storm drpc实例

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

配置

version: '2'
services:
    supervisor:
        image: storm
        container_name: supervisor
        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
        depends_on:
            - nimbus
            - zookeeper
        links:
            - nimbus
            - zookeeper
        restart: always
        ports:
            - 6700:6700
            - 6701:6701
            - 6702:6702
            - 6703:6703
            - 8000:8000
    drpc:
        image: storm
        container_name: drpc
        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
        depends_on:
            - nimbus
            - supervisor
            - zookeeper
        links:
            - nimbus
            - supervisor
            - zookeeper
        restart: always
        ports:
            - 3772:3772
            - 3773:3773
            - 3774:3774
  • 这里对supervisor配置drpc.servers及drpc.port、drpc.invocations.port,好让worker通过drpc.invocations.port去访问drpc节点
  • 对于drpc服务,则暴露drpc.port(好让外部的DRPCClient访问)、drpc.invocations.port(让worker访问)

TridentTopology

    @Test
    public void testDeployDRPCStateQuery() throws InterruptedException, TException {
        TridentTopology topology = new TridentTopology();
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 3,
                new Values("the cow jumped over the moon"),
                new Values("the man went to the store and bought some candy"),
                new Values("four score and seven years ago"),
                new Values("how many apples can you eat"));
        spout.setCycle(true);
        TridentState wordCounts =
                topology.newStream("spout1", spout)
                        .each(new Fields("sentence"), new Split(), new Fields("word"))
                        .groupBy(new Fields("word"))
                        //NOTE transforms a Stream into a TridentState object
                        .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
                        .parallelismHint(6);

        topology.newDRPCStream("words")
                .each(new Fields("args"), new Split(), new Fields("word"))
                .groupBy(new Fields("word"))
                .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"))
                .each(new Fields("count"), new FilterNull())
                .aggregate(new Fields("count"), new Sum(), new Fields("sum"));

        StormTopology stormTopology = topology.build();

        //远程提交 mvn clean package -Dmaven.test.skip=true
        //storm默认会使用System.getProperty("storm.jar")去取,如果不设定,就不能提交
        System.setProperty("storm.jar",TOPOLOGY_JAR);

        Config conf = new Config();
        conf.put(Config.NIMBUS_SEEDS,Arrays.asList("192.168.99.100")); //配置nimbus连接主机地址,比如:192.168.10.1
        conf.put(Config.NIMBUS_THRIFT_PORT,6627);//配置nimbus连接端口,默认 6627
        conf.put(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList("192.168.99.100")); //配置zookeeper连接主机地址,可以使用集合存放多个
        conf.put(Config.STORM_ZOOKEEPER_PORT,2181); //配置zookeeper连接端口,默认2181

        StormSubmitter.submitTopology("DRPCStateQuery", conf, stormTopology);
    }
  • 这里newStream创建了一个TridentState,然后newDRPCStream创建了一个DRPCStream,其stateQuery指定为前面创建的TridentState
  • 由于TridentState把结果存储到了MemoryMapState,因而这里的DRPCStream通过drpc进行stateQuery

DRPCClient

    @Test
    public void testLaunchDrpcClient() throws TException {
        Config conf = new Config();
        //NOTE 要设置Config.DRPC_THRIFT_TRANSPORT_PLUGIN属性,不然client直接跑空指针
        conf.put(Config.DRPC_THRIFT_TRANSPORT_PLUGIN,SimpleTransportPlugin.class.getName());
        conf.put(Config.STORM_NIMBUS_RETRY_TIMES,3);
        conf.put(Config.STORM_NIMBUS_RETRY_INTERVAL,10000);
        conf.put(Config.STORM_NIMBUS_RETRY_INTERVAL_CEILING,10000);
        conf.put(Config.DRPC_MAX_BUFFER_SIZE, 104857600); // 100M
        DRPCClient client = new DRPCClient(conf, "192.168.99.100", 3772);
        System.out.println(client.execute("words", "cat dog the man"));
    }
  • 注意这里的配置项不能少,否则会引发空指针
  • 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名称

doc

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏张善友的专栏

重新审视SqlDataReader的使用

      ADO.NET 1.x 利用SqlDataReader读取数据,针对每个结果集需要一个独立的连接。当然,你还必须管理这些连接并且要付出相应的内存和潜...

21690
来自专栏一个番茄说

Swift中防止ptrace依附

在移动开发中,安全是一个很重要的话题,当然安全是没有绝对的,只能说尽可能的提高安全性。在iOS的开发中,为了防止别人窥视我们的App,我们得采用一些手段来进行防...

14530
来自专栏逸鹏说道

我这么玩Web Api(一)

帮助页面或用户手册(Microsoft and Swashbuckle Help Page) 前言   你需要为客户编写Api调用手册?你需要测试你的Api接口...

32350
来自专栏技术博客

Ioc模式和MEF

  分离关注( Separation of Concerns : SOC)是Ioc模式和AOP产生最原始动力,通过功能分解可得到关注点,这些关注可以是 组件Co...

9620
来自专栏吉浦迅科技

DAY71:阅读Device-side Launch from PTX

我们正带领大家开始阅读英文的《CUDA C Programming Guide》,今天是第71天,我们正在讲解CUDA 动态并行,希望在接下来的30天里,您可以...

14220
来自专栏互联网技术栈

Netflix Archaius 分布式配置管理依赖构件

archaius是Netflix公司开源项目之一,基于java的配置管理类库,主要用于多配置存储的动态获取。主要功能是对apache common config...

17620
来自专栏微信公众号:Java团长

即将发布的 JDK 10 有 109 项新特性,你喜欢哪些?

按计划,JDK 10 将于 3 月 20 日正式发布。据前 Oracle 员工 Simon Ritter 的统计,JDK 10 总共包含 109 项新特性。当然...

8420
来自专栏Java后端技术栈

【面试题】2018年最全Java面试通关秘籍第三套!

注:本文是从众多面试者的面试经验中整理而来,其中不少是本人出的一些题目,网络资源众多,如有雷同,纯属巧合!禁止一切形式的碰瓷行为!未经允许禁止一切形式的转载和复...

14210
来自专栏java达人

java 处理xml的三种技术

最初,XML 语言仅仅是意图用来作为 HTML 语言的替代品而出现的,但是随着该语言的不断发展和完善,人们越来越发现它所具有的优点:例如标记语言可扩...

27050
来自专栏工科狗和生物喵

FreeRTOS-Qemu 实现三任务同步通信机制以及API信息

---- 1. 本次作业的考察要点: 作业地址:https://github.com/HustWolfzzb/STM32F429DiscoveryFreeRTO...

1.3K60

扫码关注云+社区

领取腾讯云代金券