DRPCSpout的outputFields为Fields("args", "return-info")，ReturnResults接收的fields为Fields("result", "return-info")

这里在SplitStreamBolt里头将stream分为两个，之后有两个CountStreamBolt分别处理两个stream的数据，最后归到同一个stream由ReportBolt消费tuple

JoinBolt继承了BaseWindowedBolt，定义了Selector selectorType、LinkedHashMap<String, JoinInfo> joinCriteria、FieldSelector[] outputFields等属性，用于记录关联类型及关联关系

storm-2.0.0/storm-client/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java

序 本文主要研究一下storm的WindowedBolt 实例 {代码...} 这里主要设置了withWindow、withTimestampField、withWatermarkInterval、withLag SlidingWindowSumBolt {代码...} TupleWindow可以获取三类值，一类是当前窗口里头的所有数据，一类是上次窗口后新到达的数据，一类是过期的数据 WindowedBolt IWindowedBolt stor...

storm-2.0.0/storm-client/src/jvm/org/apache/storm/task/IErrorReporter.java

这里对supervisor配置drpc.servers及drpc.port、drpc.invocations.port，好让worker通过drpc.invocations.port去访问drpc节点

对于bolt来说，要做两件事情，一是要anchor，在emit的时候把输入及输出tuple连接起来，构建tuple tree；而要对处理完的tuple进行ack，失败进行fail操作

这里实例是重写getComponentConfiguration方法，直接new了一个conf，设置了Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS参数

direct grouping是一种特殊的grouping，它是由上游的producer直接指定下游哪个task去接收它发射出来的tuple。direct grouping的使用有如下几个步骤：

storm-2.0.0/storm-client/src/jvm/org/apache/storm/grouping/CustomStreamGrouping.java

值得注意的是在wordCount的bolt使用PartialKeyGrouping，同一个单词不再固定发给相同的task，因此这里还需要RollingCountAggBolt按fieldsGrouping进行合并。

storm-2.0.0/storm-server/src/main/java/org/apache/storm/nimbus/AssignmentDistributionService.java

storm-2.0.0/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java

storm-core-1.2.2-sources.jar!/org/apache/storm/metrics2/reporters/GraphiteStormReporter.java

storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/worker/Worker.java

storm-core-1.2.2-sources.jar!/org/apache/storm/daemon/supervisor/Supervisor.java

curator-recipes-4.0.1-sources.jar!/org/apache/curator/framework/recipes/leader/LeaderLatch.java

序 本文主要研究一下storm nimbus的LeaderElector Nimbus org/apache/storm/daemon/nimbus/Nimbus.java {代码...} Nimbus在构造器里头调用Zookeeper.zkLeaderElector创建leaderElector launchServer方法调用了leaderElector.addToLeaderLockQueue()参与leader选举 Zookeeper.zkLeaderElector storm-core-1.1.0-sources.j...

序 本文主要研究一下storm client的nimbus.seeds参数 NIMBUS_SEEDS storm-core-1.1.0-sources.jar!/org/apache/storm/Config.java {代码...} 可以看到这里废除了nimbus.host参数，而nimbus.seeds参数主要用于发现nimbus leader StormSubmitter storm-core-1.1.0-sources.jar!/org/apache/storm/StormSubmitter.java {代...

这里可以看到这里上传到了nimbus的路径为/data/nimbus/inbox/stormjar-4ead82bb-74a3-45a3-aca4-3af2f1d23998.jar

metrics-core-4.0.2-sources.jar!/com/codahale/metrics/InstrumentedExecutorService.java

当前使用的镜像是bijukunjummen/rabbitmq-server:3.7.0，docker-compose文件配置的账号密码为myuser/mypass

序 本文主要展示一下如何使用reactor-kafka maven {代码...} 准备 启动zookeeper {代码...} 启动kafka {代码...} 创建topic {代码...} 实例 producer {代码...} consumer {代码...} 小结 reactor-kafka对kafka的api进行封装，改造为reactive streams模式，这样用起来更为顺手，熟悉reactor的开发人员可以轻车熟路。 doc ...

序 本文主要研究一下如何使用jmxtrans+influxdb+granfa监控zookeeper 配置zookeeper jmx 在conf目录下新增zookeeper-env.sh，并使用chmod +x赋予执行权限，内容如下 {代码...} 指定JMXPORT为8999 搭建influxdb及grafana 这里使用docker构建，如下 {代码...} 创建monitoring-network，让docker中的influxdb与grafana连通...

序 本文主要研究一下jdk httpclient的ConnectionPool HttpConnection HttpConnection.getConnection java.net.http/jdk/internal/net/http/HttpConnection.java {代码...} 这里非https、https1.1的，走pool.getConnection(true, addr, proxy) HttpConnection.closeOrReturnToCache java.net.http/jdk/internal/net/http/...

序 本文主要研究一下httpclient的connect timeout异常 实例代码 {代码...} 异常日志如下： {代码...} Exchange.responseAsync java.net.http/jdk/internal/net/http/Exchange.java {代码...} responseAsync最后调用ExchangeImpl.get(this, connection) ExchangeImpl.get java.net.http/jdk/internal/net/http/ExchangeIm...

另外还有一个max_attempts，读取的是jdk.httpclient.redirects.retrylimit值，读取不到默认取DEFAULT_MAX_ATTEMPTS，为5

这里如果HttpClientBuilderImpl的executor为null，则会创建Executors.newCachedThreadPool(new DefaultThreadFactory(id))

序 本文主要研究一下jdk http的HeaderFilter。 FilterFactory java.net.http/jdk/internal/net/http/FilterFactory.java {代码...} 提供了addFilter及getFilterChain方法，前者添加filter class，后者使用反射实例化filter。 HttpClientImpl java.net.http/jdk/internal/net/http/HttpClientImpl.java {代码...} HttpCli...