在Netty优化章节中介绍过可以使用线程组优化 WorkGroup 的执行效率，但在高并发环境下，应采使用 disruptor 对业务逻辑进行处理

Netty基础服务搭建：

基于Jboss编解码器案例进行修改，创建两个 SpringBoot 项目，分别为Netty客户端与服务端，服务端接收到客户端发送的Java实体类消息后创建新消息返回给客户端
image.png

Common端：

Pom依赖：

  1. <dependencies>
  2.   <dependency>
  3.     <groupId>org.projectlombok</groupId>
  4.     <artifactId>lombok</artifactId>
  5.     <version>1.18.24</version>
  6.     <scope>provided</scope>
  7.   </dependency>
  8.   <dependency>
  9.     <groupId>io.netty</groupId>
  10.     <artifactId>netty-codec</artifactId>
  11.     <version>4.1.76.Final</version>
  12.     <scope>compile</scope>
  13.   </dependency>
  14.   <!-- Netty Java序列化框架marshalling -->
  15.   <dependency>
  16.     <groupId>org.jboss.marshalling</groupId>
  17.     <artifactId>jboss-marshalling</artifactId>
  18.     <version>1.3.0.CR9</version>
  19.   </dependency>
  20.   <dependency>
  21.     <groupId>org.jboss.marshalling</groupId>
  22.     <artifactId>jboss-marshalling-serial</artifactId>
  23.     <version>1.3.0.CR9</version>
  24.   </dependency>
  25.   <dependency>
  26.     <groupId>com.lmax</groupId>
  27.     <artifactId>disruptor</artifactId>
  28.     <version>3.3.7</version>
  29.     <scope>compile</scope>
  30.   </dependency>
  31. </dependencies>

实体类：

  1. package entity;
  3. import lombok.Data;
  4. import lombok.experimental.Accessors;
  6. import java.io.Serializable;
  9. /* Netty 传输数据实体类 */
  10. @Data
  11. @Accessors(chain = true)
  12. public class TranslatorData implements Serializable {
  14.     private String id;
  15.     private String name;
  16.     private String message;     //传输消息具体内容
  18. }

Java实体类编解码器：

  1. package factory;
  3. import io.netty.handler.codec.marshalling.*;
  4. import org.jboss.marshalling.MarshallerFactory;
  5. import org.jboss.marshalling.Marshalling;
  6. import org.jboss.marshalling.MarshallingConfiguration;
  8. /**
  9.  * Marshalling工厂，Java实体类编解码器
  10.  */
  11. public final class MarshallingCodeCFactory {
  13.     /**
  14.      * 创建Jboss Marshalling解码器MarshallingDecoder
  15.      * @return MarshallingDecoder
  16.      */
  17.     public static MarshallingDecoder buildMarshallingDecoder() {
  18.         //首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。
  19.         final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
  20.         //创建了MarshallingConfiguration对象，配置了版本号为5 
  21.         final MarshallingConfiguration configuration = new MarshallingConfiguration();
  22.         configuration.setVersion(5);
  23.         //根据marshallerFactory和configuration创建provider
  24.         UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);
  25.         //构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度
  26.         MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024 * 1024 * 1);
  27.         return decoder;
  28.     }
  30.     /**
  31.      * 创建Jboss Marshalling编码器MarshallingEncoder
  32.      * @return MarshallingEncoder
  33.      */
  34.     public static MarshallingEncoder buildMarshallingEncoder() {
  35.         final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
  36.         final MarshallingConfiguration configuration = new MarshallingConfiguration();
  37.         configuration.setVersion(5);
  38.         MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration);
  39.         //构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组
  40.         MarshallingEncoder encoder = new MarshallingEncoder(provider);
  41.         return encoder;
  42.     }
  45. }

服务端：

服务端核心代码：

  1. package dmbjz.server;
  3. import factory.MarshallingCodeCFactory;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.buffer.PooledByteBufAllocator;
  6. import io.netty.channel.*;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.SocketChannel;
  9. import io.netty.channel.socket.nio.NioServerSocketChannel;
  10. import io.netty.handler.codec.marshalling.MarshallingEncoder;
  11. import io.netty.handler.codec.string.StringDecoder;
  12. import io.netty.handler.codec.string.StringEncoder;
  13. import io.netty.handler.logging.LogLevel;
  14. import io.netty.handler.logging.LoggingHandler;
  15. import lombok.extern.slf4j.Slf4j;
  16. import org.springframework.stereotype.Component;
  19. /* Netty服务端 */
  20. @Component
  21. @Slf4j
  22. public class NettyServer {
  24.     //创建单例模式
  25.     private static class SingletionWsServer{
  26.         static final NettyServer instance = new NettyServer();
  27.     }
  29.     //返回单例模式对象
  30.     public static NettyServer getInstance(){
  31.         return SingletionWsServer.instance;
  32.     }
  34.     private EventLoopGroup bossGroup;
  35.     private EventLoopGroup workGroup;
  36.     private ServerBootstrap bootstrap;
  37.     private ChannelFuture channelFuture;
  40.     /* 构造方法内进行初始化 */
  41.     public NettyServer(){
  43.         bossGroup = new NioEventLoopGroup();
  44.         workGroup = new NioEventLoopGroup();
  45.         bootstrap = new ServerBootstrap();
  47.         bootstrap.group(bossGroup,workGroup)
  48.                 .channel(NioServerSocketChannel.class)
  49.                 .option(ChannelOption.SO_BACKLOG,1024)
  50.                 //缓存区自适应,最好可以进行手动计算Bytebuf数据包大小来设置缓存区空间
  51.                 .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
  52.                 //开启日志记录
  53.                 .handler(new LoggingHandler(LogLevel.DEBUG))
  54.                 //回调方法
  55.                 .childHandler(new ChannelInitializer<SocketChannel>() {
  56.                     @Override
  57.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
  58.                         ChannelPipeline pipeline = socketChannel.pipeline();
  59.                         //添加Java实体类编解码器
  60.                         pipeline.addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
  61.                         pipeline.addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
  62.                         //添加服务端Handler
  63.                         pipeline.addLast(new ServerHandler());
  64.                     }}
  65.                 );
  67.     }
  69.     public void start(){
  70.         //额外线程启动Netty而不是在Main方法中启动，因此无需异步
  71.         //使用Spring容器进行托管，也无需进行关闭
  72.         this.channelFuture = bootstrap.bind(8765);
  73.         log.warn("Netty WebSocket Server启动完毕...");
  74.     }
  77. }

服务端Handler：

  1. package dmbjz.server;
  3. import entity.TranslatorData;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  8. /** 服务端Handler */
  9. public class ServerHandler extends ChannelInboundHandlerAdapter {
  11.     @Override
  12.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  13.         System.out.println("通道建立成功");
  14.         super.channelActive(ctx);
  15.     }
  17.     @Override
  18.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  20.         TranslatorData info = (TranslatorData) msg;
  21.         System.out.println("ID："+info.getId()+"，名称："+info.getName()+"，消息："+info.getMessage());
  23.         //创建新数据
  24.         TranslatorData response = new TranslatorData();
  25.         response.setId("resp: " + info.getId());
  26.         response.setName("resp: " + info.getName());
  27.         response.setMessage("resp: " + info.getMessage());
  29.         //返回数据
  30.         ctx.writeAndFlush(response);
  32.     }
  35. }

SpringBoot监听执行：

  1. package dmbjz;
  3. import dmbjz.server.NettyServer;
  4. import org.springframework.context.ApplicationListener;
  5. import org.springframework.context.event.ContextRefreshedEvent;
  6. import org.springframework.stereotype.Component;
  8. /* Spring加载完成后进行监听执行 */
  9. @Component
  10. public class NettyBoot implements ApplicationListener<ContextRefreshedEvent> {
  13.     /* 当一个ApplicationContext被初始化或刷新触发 */
  14.     @Override
  15.     public void onApplicationEvent(ContextRefreshedEvent event) {
  16.         //判断当前是否为spring容器初始化完成，防止重复执行
  17.         if(event.getApplicationContext().getParent() == null){
  18.             try {
  19.                 NettyServer.getInstance().start();
  20.             } catch (Exception e) {
  21.                 e.printStackTrace();
  22.             }
  23.         }
  24.     }
  26. }

客户端：

客户端代码：

  1. package com.dmbjz.client;
  3. import entity.TranslatorData;
  4. import factory.MarshallingCodeCFactory;
  5. import io.netty.bootstrap.Bootstrap;
  6. import io.netty.channel.*;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.SocketChannel;
  9. import io.netty.channel.socket.nio.NioSocketChannel;
  10. import io.netty.handler.logging.LogLevel;
  11. import io.netty.handler.logging.LoggingHandler;
  14. public class NettyClient {
  16.     private Channel channel;
  18.     public NettyClient() {
  19.         this.connect();
  20.     }
  22.     private void connect() {
  24.         EventLoopGroup workGroup = new NioEventLoopGroup();
  25.         Bootstrap bootstrap = new Bootstrap();
  26.         try {
  28.             bootstrap.group(workGroup)
  29.             .channel(NioSocketChannel.class)
  30.             //表示缓存区动态调配（自适应）
  31.             .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
  32.             .handler(new LoggingHandler(LogLevel.INFO))
  33.             .handler(new ChannelInitializer<SocketChannel>() {
  34.                 @Override
  35.                 protected void initChannel(SocketChannel sc) throws Exception {
  36.                     ChannelPipeline pipeline = sc.pipeline();
  37.                     //添加Java实体类编解码器
  38.                     pipeline.addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
  39.                     pipeline.addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
  40.                     //添加自定义处理器
  41.                     pipeline.addLast(new ClientHandler());
  42.                 }
  43.             });
  45.             //绑定端口，同步等等请求连接
  46.             ChannelFuture future = bootstrap.connect("127.0.0.1", 8765).sync();
  47.             System.out.println("客户端开始连接....");
  49.             //进行数据的发送, 但是首先我们要获取channel:
  50.             this.channel = future.channel();
  52.         } catch (InterruptedException e) {
  53.             e.printStackTrace();
  54.         }
  56.     }
  59.     /* 数据发送方法 */
  60.     public void sendData(){
  62.         for(int i =0; i <10; i++){
  63.             TranslatorData request = new TranslatorData();
  64.             request.setId("" + i);
  65.             request.setName("请求消息名称 " + i);
  66.             request.setMessage("请求消息内容 " + i);
  67.             this.channel.writeAndFlush(request);
  68.         }
  70.     }
  74. }

客户端Handler：

  1. package com.dmbjz.client;
  3. import entity.TranslatorData;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  6. import io.netty.util.ReferenceCountUtil;
  9. public class ClientHandler extends ChannelInboundHandlerAdapter {
  11.     @Override
  12.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  14.         try {
  15.             TranslatorData response = (TranslatorData)msg;
  16.             System.err.println("Client端: id= " + response.getId() 
  17.                     + ", name= " + response.getName()
  18.                     + ", message= " + response.getMessage());
  19.         } finally {
  20.             //释放缓存
  21.             ReferenceCountUtil.release(msg);
  22.         }
  24.     }
  26. }

测试：

image.png
image.png

整合 Disruptor：

每个用户创建出唯一ID，将服务端与消费端需要处理的业务逻辑交给 disruptor 的生产者，消费者通过唯一ID获取到用户在客户端传递的数据进行业务处理
Netty整合Disruptor - 图4
image.png

Common端：

消费者：

  1. package disruptor;
  3. import com.lmax.disruptor.WorkHandler;
  6. /* 消费者 */
  7. public abstract class MessageConsumer implements WorkHandler<TranslatorDataWrapper> {
  9.     protected String consumerId;
  11.     public MessageConsumer(String consumerId) {
  12.         this.consumerId = consumerId;
  13.     }
  15.     public String getConsumerId() {
  16.         return consumerId;
  17.     }
  19.     public void setConsumerId(String consumerId) {
  20.         this.consumerId = consumerId;
  21.     }
  23. }

生产者：

  1. package disruptor;
  3. import com.lmax.disruptor.RingBuffer;
  4. import entity.TranslatorData;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import lombok.AllArgsConstructor;
  7. import lombok.Getter;
  8. import lombok.NoArgsConstructor;
  9. import lombok.Setter;
  12. /* 生产者 */
  13. @NoArgsConstructor
  14. @AllArgsConstructor
  15. @Getter
  16. @Setter
  17. public class MessageProductor {
  19.     private String producerId;
  20.     private RingBuffer<TranslatorDataWrapper> ringBuffer;
  23.     /* 发布数据到 RingBuffer */
  24.     public void onData(TranslatorData data, ChannelHandlerContext ctx) {
  25.         long sequence = ringBuffer.next();
  26.         try {
  27.             TranslatorDataWrapper wapper = ringBuffer.get(sequence);
  28.             wapper.setTranslatorData(data);
  29.             wapper.setCtx(ctx);
  30.         } finally {
  31.             ringBuffer.publish(sequence);
  32.         }
  33.     }
  35. }

传输实体类：

  1. package disruptor;
  3. import entity.TranslatorData;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import lombok.Data;
  6. import lombok.NoArgsConstructor;
  9. /* Disruptor 传递的数据实体类 */
  10. @Data
  11. @NoArgsConstructor
  12. public class TranslatorDataWrapper {
  14.     private TranslatorData translatorData;      //数据实体类
  15.     private ChannelHandlerContext ctx;          //Netty使用的Ctx对象
  17. }

失败操作：

  1. package disruptor;
  4. import com.lmax.disruptor.ExceptionHandler;
  7. /* 事件处理失败时的操作 */
  8. public class EventExceptionHandler implements ExceptionHandler<TranslatorDataWrapper> {
  10.     @Override
  11.     public void handleEventException(Throwable ex, long sequence, TranslatorDataWrapper event) {
  12.         System.out.println("消费时出现异常");
  13.     }
  15.     @Override
  16.     public void handleOnStartException(Throwable ex) {
  17.         System.out.println("启动时出现异常");
  18.     }
  20.     @Override
  21.     public void handleOnShutdownException(Throwable ex) {
  22.         System.out.println("关闭时出现异常");
  23.     }
  25. }

Disruptor实现：

  1. package disruptor;
  4. import com.lmax.disruptor.*;
  5. import com.lmax.disruptor.dsl.ProducerType;
  7. import java.util.Map;
  8. import java.util.concurrent.ArrayBlockingQueue;
  9. import java.util.concurrent.ConcurrentHashMap;
  10. import java.util.concurrent.ThreadPoolExecutor;
  11. import java.util.concurrent.TimeUnit;
  13. /* Disruptor实现 */
  14. public class RingBufferWorkerPoolFactory {
  17.     private static class SingletonHolder {
  18.         static final RingBufferWorkerPoolFactory instance = new RingBufferWorkerPoolFactory();
  19.     }
  21.     //返回单例模式对象
  22.     public static RingBufferWorkerPoolFactory getInstance(){
  23.         return SingletonHolder.instance;
  24.     }
  26.     private RingBufferWorkerPoolFactory(){
  28.     }
  30.     //用Map管理生产者
  31.     private static Map<String,MessageProductor> productorMap = new ConcurrentHashMap<String, MessageProductor>();
  33.     //用Map管理消费者
  34.     private static Map<String,MessageConsumer> consumerMap = new ConcurrentHashMap<String, MessageConsumer>();
  36.     //序号栅栏
  37.     private SequenceBarrier sequenceBarrier;
  38.     //环形数组
  39.     private RingBuffer<TranslatorDataWrapper> ringBuffer;
  40.     //工作池
  41.     private WorkerPool<TranslatorDataWrapper> workerPool;
  44.     /*
  45.      * 初始化Disruptor并启动
  46.      * @param type生产者类型
  47.      * @param bufferSize 环形缓冲区中创建的元素数
  48.      * @param waitStrategy等待策略
  49.      * @param messageConsumers消费者数组
  50.      */
  51.     public void initAndStart(ProducerType type, int bufferSize, WaitStrategy waitStrategy, MessageConsumer[] messageConsumers) {
  53.         ThreadPoolExecutor threadPool = new ThreadPoolExecutor(8,16,10, TimeUnit.SECONDS,new ArrayBlockingQueue<>(20));
  55.         ringBuffer = RingBuffer.create(type,()->new TranslatorDataWrapper(),bufferSize,new YieldingWaitStrategy());
  57.         //设置序号栅栏(创建序号屏障)
  58.         sequenceBarrier =  ringBuffer.newBarrier();
  60.         //构建多消费者工作池
  61.         workerPool = new WorkerPool<TranslatorDataWrapper>(ringBuffer,sequenceBarrier,new EventExceptionHandler(),messageConsumers);
  63.         //把构建的消费者放入池中
  64.         for (MessageConsumer consumer : messageConsumers) {
  65.             consumerMap.put(consumer.getConsumerId(), consumer);
  66.         }
  68.         //每个消费者的Sequence序号都是单独的，通过WorkPool获取每个消费者的序号然后设置到RingBuffer中
  69.         ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
  71.         //启动工作池
  72.         workerPool.start(threadPool);
  74.     }
  77.     /* 通过ID获取消费者 */
  78.     public MessageProductor getMessageProducer(String producerId){
  79.         MessageProductor messageProducer = productorMap.get(producerId);
  80.         if(null == messageProducer) {
  81.             messageProducer = new MessageProductor(producerId,ringBuffer);
  82.             productorMap.put(producerId, messageProducer);
  83.         }
  84.         return messageProducer;
  85.     }
  88. }

服务端：

添加消费者：

  1. package dmbjz.server;
  3. import disruptor.MessageConsumer;
  4. import disruptor.TranslatorDataWrapper;
  5. import entity.TranslatorData;
  6. import io.netty.channel.ChannelHandlerContext;
  8. public class MessageConsumerImplServer extends MessageConsumer {
  10.     public MessageConsumerImplServer(String consumerId) {
  11.         super(consumerId);
  12.     }
  14.     @Override
  15.     public void onEvent(TranslatorDataWrapper event) throws Exception {
  17.         TranslatorData request = event.getTranslatorData();
  18.         ChannelHandlerContext ctx = event.getCtx();
  20.         //业务处理逻辑:
  21.         System.out.println("Sever端: id= " + request.getId() + ", name= " + request.getName() + ", message= " + request.getMessage());
  23.         //回送响应信息:
  24.         TranslatorData response = new TranslatorData();
  25.         response.setId("resp: " + request.getId());
  26.         response.setName("resp: " + request.getName());
  27.         response.setMessage("resp: " + request.getMessage());
  29.         //写出response响应信息:
  30.         ctx.writeAndFlush(response);
  32.     }
  34. }

修改ServcerHandler：

使用 disruptor 管理业务处理逻辑

  1. package dmbjz.server;
  3. import disruptor.MessageProductor;
  4. import disruptor.RingBufferWorkerPoolFactory;
  5. import entity.TranslatorData;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.channel.ChannelInboundHandlerAdapter;
  10. /** 服务端Handler */
  11. public class ServerHandler extends ChannelInboundHandlerAdapter {
  13.     @Override
  14.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  15.         System.out.println("通道建立成功");
  16.         super.channelActive(ctx);
  17.     }
  19.     @Override
  20.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  22.         /*
  23.             TranslatorData info = (TranslatorData) msg;
  24.             System.out.println("ID："+info.getId()+"，名称："+info.getName()+"，消息："+info.getMessage());
  26.             //创建新数据
  27.             TranslatorData response = new TranslatorData();
  28.             response.setId("resp: " + info.getId());
  29.             response.setName("resp: " + info.getName());
  30.             response.setMessage("resp: " + info.getMessage());
  32.             //返回数据
  33.             ctx.writeAndFlush(response);
  35.          */
  37.         /* 使用disruptor异步执行业务逻辑，将数据发送给消费者即可 */
  38.         TranslatorData info = (TranslatorData) msg;
  39.         String id = "code:seesionId:001";               //同一个用户应使用同一个Id，例如可以使用机器码:sessionId:用户编号作为规则进行生成，这里固定写死
  40.         MessageProductor messageProducer = RingBufferWorkerPoolFactory.getInstance().getMessageProducer(id);
  41.         messageProducer.onData(info,ctx);
  43.     }
  46. }

修改NettyBoot：

  1. package dmbjz;
  3. import com.lmax.disruptor.BlockingWaitStrategy;
  4. import com.lmax.disruptor.YieldingWaitStrategy;
  5. import com.lmax.disruptor.dsl.ProducerType;
  6. import disruptor.MessageConsumer;
  7. import disruptor.RingBufferWorkerPoolFactory;
  8. import dmbjz.server.MessageConsumerImplServer;
  9. import dmbjz.server.NettyServer;
  10. import org.springframework.context.ApplicationListener;
  11. import org.springframework.context.event.ContextRefreshedEvent;
  12. import org.springframework.stereotype.Component;
  14. /* Spring加载完成后进行监听执行 */
  15. @Component
  16. public class NettyBoot implements ApplicationListener<ContextRefreshedEvent> {
  19.     /* 当一个ApplicationContext被初始化或刷新触发 */
  20.     @Override
  21.     public void onApplicationEvent(ContextRefreshedEvent event) {
  23.         //判断当前是否为spring容器初始化完成，防止重复执行
  24.         if(event.getApplicationContext().getParent() == null){
  25.             try {
  27.                 //初始化消费者
  28.                 MessageConsumer[] conusmers = new MessageConsumer[4];
  29.                 for(int i =0; i < conusmers.length; i++) {
  30.                     MessageConsumer messageConsumer = new MessageConsumerImplServer("code:serverId:" + i);
  31.                     conusmers[i] = messageConsumer;
  32.                 }
  33.                 //初始化Disruptor并启动
  34.                 RingBufferWorkerPoolFactory.getInstance().initAndStart(
  35.                         ProducerType.MULTI,
  36.                         1024*1024,
  37.                         //new YieldingWaitStrategy(),
  38.                         new BlockingWaitStrategy(),
  39.                         conusmers);
  41.                 //启动Netty服务端
  42.                 NettyServer.getInstance().start();
  44.             } catch (Exception e) {
  45.                 e.printStackTrace();
  46.             }
  47.         }
  49.     }
  51. }

修改客户端：

修改ClientHandler：

  1. package com.dmbjz.client;
  3. import disruptor.MessageProductor;
  4. import disruptor.RingBufferWorkerPoolFactory;
  5. import entity.TranslatorData;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.channel.ChannelInboundHandlerAdapter;
  8. import io.netty.util.ReferenceCountUtil;
  11. public class ClientHandler extends ChannelInboundHandlerAdapter {
  13.     @Override
  14.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  16.         /*
  17.                 try {
  18.                     TranslatorData response = (TranslatorData)msg;
  19.                     System.err.println("Client端: id= " + response.getId()
  20.                             + ", name= " + response.getName()
  21.                             + ", message= " + response.getMessage());
  22.                 } finally {
  23.                     //释放缓存
  24.                     ReferenceCountUtil.release(msg);
  25.                 }
  26.         */
  28.         TranslatorData response = (TranslatorData)msg;
  29.         String producerId = "code:seesionId:002";
  30.         MessageProductor messageProducer = RingBufferWorkerPoolFactory.getInstance().getMessageProducer(producerId);
  31.         messageProducer.onData(response, ctx);
  34.     }
  36. }

添加消费者：

  1. package com.dmbjz.client;
  3. import disruptor.MessageConsumer;
  4. import disruptor.TranslatorDataWrapper;
  5. import entity.TranslatorData;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.util.ReferenceCountUtil;
  10. public class MessageConsumerImplClient extends MessageConsumer {
  12.     public MessageConsumerImplClient(String consumerId) {
  13.         super(consumerId);
  14.     }
  17.     /* 业务逻辑处理方式 */
  18.     @Override
  19.     public void onEvent(TranslatorDataWrapper event) throws Exception {
  21.         TranslatorData response = event.getTranslatorData();
  22.         //业务逻辑处理:
  23.         try {
  24.             System.out.println("Client端: id= " + response.getId() + ", name= " + response.getName() + ", message= " + response.getMessage());
  25.         } finally {
  26.             ReferenceCountUtil.release(response);
  27.         }
  29.     }
  31. }

修改启动类：

  1. package com.dmbjz;
  3. import com.dmbjz.client.MessageConsumerImplClient;
  4. import com.dmbjz.client.NettyClient;
  5. import com.lmax.disruptor.BlockingWaitStrategy;
  6. import com.lmax.disruptor.dsl.ProducerType;
  7. import disruptor.MessageConsumer;
  8. import disruptor.RingBufferWorkerPoolFactory;
  9. import org.springframework.boot.SpringApplication;
  10. import org.springframework.boot.autoconfigure.SpringBootApplication;
  12. @SpringBootApplication
  13. public class NettyClientApplication {
  15.     public static void main(String[] args) {
  17.         SpringApplication.run(NettyClientApplication.class, args);
  19.         //创建消费者
  20.         MessageConsumer[] conusmers = new MessageConsumer[4];
  21.         for(int i =0; i < conusmers.length; i++) {
  22.             MessageConsumer messageConsumer = new MessageConsumerImplClient("code:clientId:" + i);
  23.             conusmers[i] = messageConsumer;
  24.         }
  26.         //初始化并启动disruptor
  27.         RingBufferWorkerPoolFactory.getInstance().initAndStart(ProducerType.MULTI,
  28.                 1024*1024,
  29.                 //new YieldingWaitStrategy(),
  30.                 new BlockingWaitStrategy(),
  31.                 conusmers);
  33.         //在启动的时候执行Client代码
  34.         new NettyClient().sendData();
  36.     }
  38. }

测试：

image.png image.png