storm drpc实例

序

本文主要演示一下storm drpc实例

配置

1version: '2'
2services:
3    supervisor:
4        image: storm
5        container_name: supervisor
6        command: storm supervisor -c storm.local.hostname="192.168.99.100" -c drpc.servers='["192.168.99.100"]' -c drpc.port=3772 -c drpc.invocations.port=3773 -c drpc.http.port=3774
7        depends_on:
8            - nimbus
9            - zookeeper
10        links:
11            - nimbus
12            - zookeeper
13        restart: always
14        ports:
15            - 6700:6700
16            - 6701:6701
17            - 6702:6702
18            - 6703:6703
19            - 8000:8000
20    drpc:
21        image: storm
22        container_name: drpc
23        command: storm drpc -c storm.local.hostname="192.168.99.100" -c drpc.port=3772 -c drpc.invocations.port=3773 -c drpc.http.port=3774
24        depends_on:
25            - nimbus
26            - supervisor
27            - zookeeper
28        links:
29            - nimbus
30            - supervisor
31            - zookeeper
32        restart: always
33        ports:
34            - 3772:3772
35            - 3773:3773
36            - 3774:3774

这里对supervisor配置drpc.servers及drpc.port、drpc.invocations.port，好让worker通过drpc.invocations.port去访问drpc节点
对于drpc服务，则暴露drpc.port(好让外部的DRPCClient访问)、drpc.invocations.port(让worker访问)

TridentTopology

1    @Test
2    public void testDeployDRPCStateQuery() throws InterruptedException, TException {
3        TridentTopology topology = new TridentTopology();
4        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 3,
5                new Values("the cow jumped over the moon"),
6                new Values("the man went to the store and bought some candy"),
7                new Values("four score and seven years ago"),
8                new Values("how many apples can you eat"));
9        spout.setCycle(true);
10        TridentState wordCounts =
11                topology.newStream("spout1", spout)
12                        .each(new Fields("sentence"), new Split(), new Fields("word"))
13                        .groupBy(new Fields("word"))
14                        //NOTE transforms a Stream into a TridentState object
15                        .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
16                        .parallelismHint(6);
17
18        topology.newDRPCStream("words")
19                .each(new Fields("args"), new Split(), new Fields("word"))
20                .groupBy(new Fields("word"))
21                .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"))
22                .each(new Fields("count"), new FilterNull())
23                .aggregate(new Fields("count"), new Sum(), new Fields("sum"));
24
25        StormTopology stormTopology = topology.build();
26
27        //远程提交 mvn clean package -Dmaven.test.skip=true
28        //storm默认会使用System.getProperty("storm.jar")去取，如果不设定，就不能提交
29        System.setProperty("storm.jar",TOPOLOGY_JAR);
30
31        Config conf = new Config();
32        conf.put(Config.NIMBUS_SEEDS,Arrays.asList("192.168.99.100")); //配置nimbus连接主机地址，比如：192.168.10.1
33        conf.put(Config.NIMBUS_THRIFT_PORT,6627);//配置nimbus连接端口，默认 6627
34        conf.put(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList("192.168.99.100")); //配置zookeeper连接主机地址，可以使用集合存放多个
35        conf.put(Config.STORM_ZOOKEEPER_PORT,2181); //配置zookeeper连接端口，默认2181
36
37        StormSubmitter.submitTopology("DRPCStateQuery", conf, stormTopology);
38    }

这里newStream创建了一个TridentState，然后newDRPCStream创建了一个DRPCStream，其stateQuery指定为前面创建的TridentState
由于TridentState把结果存储到了MemoryMapState，因而这里的DRPCStream通过drpc进行stateQuery

DRPCClient

1    @Test
2    public void testLaunchDrpcClient() throws TException {
3        Config conf = new Config();
4        //NOTE 要设置Config.DRPC_THRIFT_TRANSPORT_PLUGIN属性，不然client直接跑空指针
5        conf.put(Config.DRPC_THRIFT_TRANSPORT_PLUGIN,SimpleTransportPlugin.class.getName());
6        conf.put(Config.STORM_NIMBUS_RETRY_TIMES,3);
7        conf.put(Config.STORM_NIMBUS_RETRY_INTERVAL,10000);
8        conf.put(Config.STORM_NIMBUS_RETRY_INTERVAL_CEILING,10000);
9        conf.put(Config.DRPC_MAX_BUFFER_SIZE, 104857600); // 100M
10        DRPCClient client = new DRPCClient(conf, "192.168.99.100", 3772);
11        System.out.println(client.execute("words", "cat dog the man"));
12    }

注意这里的配置项不能少，否则会引发空指针
Config.DRPC_THRIFT_TRANSPORT_PLUGIN这里使用的是SimpleTransportPlugin.class.getName()，虽然该类被废弃了，不过还可以跑通
由于使用了SimpleTransportPlugin.class，因而这里要配置Config.DRPC_MAX_BUFFER_SIZE
DRPCClient配置了drpc的地址及port
client.execute这里要传入newDRPCStream指定的function名称

小结

使用drpc的时候，需要通过storm drpc启动drpc server服务节点，另外要暴露两个端口，一个drpc.port是供外部DRPCClient调用，一个drpc.invocations.port是给worker来访问；drpc.http.port端口是暴露给http协议调用的(DRPCClient使用的是thrift协议调用)
supervisor要配置drpc.servers、drpc.invocations.port，好让worker去访问到drpc server
DRPCClient使用drpc.port指定的端口来访问，另外client.execute这里要传入newDRPCStream指定的function名称

序

配置

TridentTopology

DRPCClient

小结

doc

评论区

相关推荐