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...

从java9的jdk.incubator.httpclient模块迁移到java.net.http模块，包名由jdk.incubator.http改为java.net.http

序 本文主要研究一下Java Flight Recorder的使用。 命令 主要有5个命令，configure、check、start、dump、stop。执行顺序的话，先start再dump，最后stop。 JFR.configure 参数 描述 值类型 默认值 globalbuffercount 指定global buffers的数量. 修改 memorysize参数会影响该值. Long 默认值依赖 memorysize 参数. global...

Java语言特性系列Java5的新特性Java6的新特性Java7的新特性Java8的新特性Java9的新特性Java10的新特性Java11的新特性Java12的新特性Java13的新特性Java14的新特性Java15的新特性Java16的新特性Java17的新特性Java18的新特性Java19的新特性Java20的新特性Java21的新特性Java22的新特性序本文主要讲述一下Java11的新特性版...

序 本文主要研究一下redisson的RMap的computeIfAbsent操作 实例 {代码...} 源码分析 ConcurrentMap.computeIfAbsent java/util/concurrent/ConcurrentMap.java {代码...} computeIfAbsent当该key不存在时，返回的是新值，而非null computeIfAbsent方法里头调用了putIfAbsent RedissonMap.putIfAbsent redisson-3.8.1-sou...

序 本文主要研究一下如何在springboot工程整合vue maven {代码...} 新建springboot的web工程，默认会在resources目录下生成static以及templates文件夹 templates文件用于存放后端渲染的模板，这里我们采用前后端分离的方式，因而该文件夹就没有用了 static文件夹就是存放静态文件的地方 plugin {代码...} 这里我们使用了...

序 本文主要研究一下redis的监控工具 redis-stat redis-stat是一个比较有名的redis指标可视化的监控工具，采用ruby开发，基于redis的info命令来统计，不影响redis性能。 docker运行 {代码...} 运行实例图 RedisLive RedisLive是采用python开发的redis的可视化及查询分析工具 docker运行 {代码...} 运行实例图 访问[链接]...

这里使用了两个queue，对delayedQueue的offer操作是直接进入delayedQueue，但是delay是作用在目标队列上，这里就是RBlockingQueue

这里leaseTime没有设置的话，默认是-1，使用的是commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()，默认为30秒

序 本文主要研究一下redis的数据结构的应用 string 最常用的就是incr操作，比如可以用来维护用户在某个抽奖活动的剩余抽奖次数 setnx方法可以用来实现分布式锁 hashmap 可以用来存储session，作为分布式session的一个实现方案 可以用来存储用户购物车，value值存储的key为物品，value为其数量 set set可以用来存储每个标...