14.7 DRPC

深藏阁楼爱情的钟 2022-05-12 01:14 126阅读 0赞

什么是DRPC: --是分布式远程调用

– RPC（Remote Procedure Call Protocol）——远程过程调用协议

– Distributed RPC：rpc请求流式、并进行处理

– RPC请求参数当做输入流，结果当做输出流

– 利用storm的分布式进行处理机制和能力

– 借助DRPC server接收请求、返回相应

Storm只能获取数据，不能接请求和发响应，所以这里借助一个DRPC Server来帮 助完成

DRPC把大量请求分布式的去做，一次请求如果串行的话可能会比较慢，我并行的来 处理，另一方面通过来降低平均一次请求的时间，解决了响应的吞吐

配置DRPC：apache-storm-0.9.5/conf/storm.yaml

修改drpc.servers配置，同步到其他节点

drpc.servers:

\- "node2"

\#启动DRPC server:

1，首先启动storm集群

, 2，然后：./bin/storm drpc >> ./logs/drpc.out 2>&1 & 在主节点上启动就行

--------------------

在集群创建DRPC：给这个方法打jar包，放到storm集群中运行：

./bin/storm jar /opt/local/drpc.jar com.bjsxt.basic.drpc.ManualDRPC drpc

package com.bjsxt.basic.drpc;

import backtype.storm.Config;

import backtype.storm.LocalCluster;

import backtype.storm.LocalDRPC;

import backtype.storm.StormSubmitter;

import backtype.storm.drpc.DRPCSpout;

import backtype.storm.drpc.ReturnResults;

import backtype.storm.generated.AlreadyAliveException;

import backtype.storm.generated.InvalidTopologyException;

import backtype.storm.topology.BasicOutputCollector;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.TopologyBuilder;

import backtype.storm.topology.base.BaseBasicBolt;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Tuple;

import backtype.storm.tuple.Values;

/\*\*

\* 在集群创建DRPC

\* @author root

\*/

public class ManualDRPC \{

public static class ExclamationBolt extends BaseBasicBolt \{

@Override

public void declareOutputFields(OutputFieldsDeclarer declarer) \{

declarer.declare(new Fields("result", "return-info"));

//可以写自己的逻辑

@Override

public void execute(Tuple tuple, BasicOutputCollector collector) \{

String arg = tuple.getString(0);

Object retInfo = tuple.getValue(1);

collector.emit(new Values(arg + "!!!", retInfo));

public static void main(String\[\] args) \{

TopologyBuilder builder = new TopologyBuilder();

LocalDRPC drpc = new LocalDRPC();

if (args.length > 0) \{

DRPCSpout spout = new DRPCSpout("exclamation");

builder.setSpout("drpc", spout);

builder.setBolt("exclaim", new ExclamationBolt(), 3)

.shuffleGrouping("drpc");

builder.setBolt("return", new ReturnResults(), 3).shuffleGrouping(

"exclaim");

Config conf = new Config();

try \{

StormSubmitter.submitTopology(args\[0\], conf,

builder.createTopology());

\} catch (AlreadyAliveException e) \{

e.printStackTrace();

\} catch (InvalidTopologyException e) \{

e.printStackTrace();

\} else \{

DRPCSpout spout = new DRPCSpout("exclamation", drpc);

builder.setSpout("drpc", spout);

builder.setBolt("exclaim", new ExclamationBolt(), 3)

.shuffleGrouping("drpc");

builder.setBolt("return", new ReturnResults(), 3).shuffleGrouping(

"exclaim");

LocalCluster cluster = new LocalCluster();

Config conf = new Config();

cluster.submitTopology("exclaim", conf, builder.createTopology());

--------------------

Java创建客户端连接DRPC：DRPC在集群上创建之后，eclipse可以运行

package com.bjsxt.basic.drpc;

import org.apache.thrift7.TException;

import backtype.storm.generated.DRPCExecutionException;

import backtype.storm.utils.DRPCClient;

/\*\*

\* 创建客户端访问DRPC

\* @author root

\*/

public class MyDRPCclient \{

/\*\*

\* @param args

\*/

public static void main(String\[\] args) \{

DRPCClient client = new DRPCClient("node2", 3772);

try \{

String result = client.execute("exclamation", "hello ");

System.out.println(result);

\} catch (TException e) \{

e.printStackTrace();

\} catch (DRPCExecutionException e) \{

e.printStackTrace();

14.7 DRPC

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 LeetCode Weekly Contest 147

相关 147-详细地学习指针

相关 LeetCode147—Insertion Sort List

相关 LeetCode147—Insertion Sort List

相关 [leetcode-排序]--147. Insertion Sort List

相关 Storm: DRPC简介

相关 14.7 DRPC

相关 error: #147: declaration is incompatible with ……

相关 storm drpc分布式本地和远程调用模式讲解

相关 storm manual drpc 的远程调用