首页 > 其他分享 >手写MQ消息中间件（基于netty实现）

手写MQ消息中间件（基于netty实现）

时间：2022-12-28 22:00:52浏览次数：36  
标签：netty mayikt new MQ io 消息中间件 import channel

通讯协议，什么是协议？

如：生产者投递消息到mq服务器之间的通讯如何定义格式就叫协议，（mq协议是amqp协议，是mq作者自己起的通讯协议名称）。

流程图

实现流程:

1.生产者把消息投递到netty服务，消费者自己取。

2.生产者把消息投递到netty服务，自动发送给消费者。

目录结构

目录

流程图

目录结构

MarshallingCodeCFactory

NettyMQServer

NettyMQProducer

NettyMQConsumer

NettyMQServerHandler

NettyMQProducerHandler

NettyMQConsumerHandler

DeliveryInfoEntity

MarshallingCodeCFactory

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

NettyMQServer

netty服务器 - MQ服务器
  1. package com.mayikt.mq;
  2.  
  3. import com.mayikt.handler.NettyMQServerHandler;
  4. import com.mayikt.utils.MarshallingCodeCFactory;
  5. import io.netty.bootstrap.ServerBootstrap;
  6. import io.netty.channel.ChannelFuture;
  7. import io.netty.channel.ChannelInitializer;
  8. import io.netty.channel.nio.NioEventLoopGroup;
  9. import io.netty.channel.socket.SocketChannel;
  10. import io.netty.channel.socket.nio.NioServerSocketChannel;
  11.  
  12. /**
  13.  * netty服务器 - MQ服务器
  14.  */
  15. public class NettyMQServer {
  16.  
  17.     public static void main(String[] args) {
  18.         start(5872);
  19.     }
  20.  
  21.     public static void start(int port) {
  22.         /**
  23.          *  客户端创建两个线程池组分别为 boss线程组和工作线程组
  24.          */
  25.         // 1.用于接受客户端连接的请求 （并没有处理请求）
  26.         NioEventLoopGroup bossGroup = new NioEventLoopGroup();
  27.         // 2.用于处理客户端连接的读写操作
  28.         NioEventLoopGroup workGroup = new NioEventLoopGroup();
  29.         // 3.用于创建我们的ServerBootstrap
  30.         ServerBootstrap serverBootstrap = new ServerBootstrap();
  31.         serverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class)
  32.                 .childHandler(new ChannelInitializer<SocketChannel>() {
  33.                     @Override
  34.                     protected void initChannel(SocketChannel ch) throws Exception {
  35.                         // 解决netty可以支持传输对象
  36.                         ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
  37.                         ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
  38.                         ch.pipeline().addLast(new NettyMQServerHandler());
  39.  
  40.                     }
  41.                 });
  42.         //  绑定我们的端口号码
  43.         try {
  44.             // 绑定端口号，同步等待成功
  45.             ChannelFuture future = serverBootstrap.bind(port).sync();
  46.             System.out.println("MQ服务器启动成功：" + port);
  47.             // 等待服务器监听端口
  48.             future.channel().closeFuture().sync();
  49.         } catch (Exception e) {
  50.             e.printStackTrace();
  51.  
  52.         } finally {
  53.             // 优雅的关闭连接
  54.             bossGroup.shutdownGracefully();
  55.             workGroup.shutdownGracefully();
  56.         }
  57.     }
  58.  
  59.  
  60. }

NettyMQProducer

netty客户端 - 生产者投递消息

  1. package com.mayikt.mq;
  2. import com.mayikt.entity.DeliveryInfoEntity;
  3. import com.mayikt.handler.NettyMQProducerHandler;
  4. import com.mayikt.handler.NettyMQServerHandler;
  5. import com.mayikt.utils.MarshallingCodeCFactory;
  6. import io.netty.bootstrap.Bootstrap;
  7. import io.netty.bootstrap.ServerBootstrap;
  8. import io.netty.channel.ChannelFuture;
  9. import io.netty.channel.ChannelInitializer;
  10. import io.netty.channel.nio.NioEventLoopGroup;
  11. import io.netty.channel.socket.SocketChannel;
  12. import io.netty.channel.socket.nio.NioServerSocketChannel;
  13. import io.netty.channel.socket.nio.NioSocketChannel;
  14.  
  15. import java.net.InetSocketAddress;
  16.  
  17. /**
  18.  * netty客户端 - 消费者
  19.  */
  20. public class NettyMQProducer {
  21.  
  22.     // netty服务端ip、端口
  23.     private static final String host = "127.0.0.1";
  24.     private static final int port = 5872;
  25.  
  26.     public static void main(String[] args) {
  27.         sendMsg("mayikt", "每特123");
  28.         System.out.println("111");
  29.         sendMsg("mayikt", "每特123");
  30.     }
  31.  
  32.     public static void sendMsg(String queueName, String msg) {
  33.         //创建nioEventLoopGroup
  34.         NioEventLoopGroup group = new NioEventLoopGroup();
  35.         Bootstrap bootstrap = new Bootstrap();
  36.         bootstrap.group(group).channel(NioSocketChannel.class)
  37.                 .remoteAddress(new InetSocketAddress(host, port))
  38.                 .handler(new ChannelInitializer<SocketChannel>() {
  39.                     @Override
  40.                     protected void initChannel(SocketChannel ch) throws Exception {
  41.                         ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
  42.                         ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
  43.                         ch.pipeline().addLast(new NettyMQProducerHandler());
  44.                     }
  45.                 });
  46.         try {
  47.             // 发起同步连接
  48.             ChannelFuture sync = bootstrap.connect().sync();
  49.             DeliveryInfoEntity deliveryInfoEntity = new DeliveryInfoEntity(msg, queueName,
  50.                     true);
  51.             // 发送数据给netty服务端
  52.             sync.channel().writeAndFlush(deliveryInfoEntity);
  53.             sync.channel().closeFuture().sync();
  54.         } catch (Exception e) {
  55.  
  56.         } finally {
  57.             group.shutdownGracefully();
  58.         }
  59.     }
  60.  
  61. }

NettyMQConsumer

netty客户端 - 消费者接收指定队列消息

  1. package com.mayikt.mq;
  2.  
  3. import com.mayikt.entity.DeliveryInfoEntity;
  4. import com.mayikt.handler.NettyMQConsumerHandler;
  5. import com.mayikt.utils.MarshallingCodeCFactory;
  6. import io.netty.bootstrap.Bootstrap;
  7. import io.netty.channel.ChannelFuture;
  8. import io.netty.channel.ChannelInitializer;
  9. import io.netty.channel.nio.NioEventLoopGroup;
  10. import io.netty.channel.socket.SocketChannel;
  11. import io.netty.channel.socket.nio.NioSocketChannel;
  12.  
  13. import java.net.InetSocketAddress;
  14.  
  15. /**
  16.  * netty客户端 - 生产者投递消息
  17.  */
  18. public class NettyMQConsumer {
  19.  
  20.     // netty服务端ip、端口
  21.     private static final String host = "127.0.0.1";
  22.     private static final int port = 5872;
  23.  
  24.     private static String queueName = "mayikt";
  25.  
  26.     public static void main(String[] args) {
  27.         sendMsg("mayikt",null);
  28.     }
  29.  
  30.     public static void sendMsg(String queueName,String msg) {
  31.         //创建nioEventLoopGroup
  32.         NioEventLoopGroup group = new NioEventLoopGroup();
  33.         Bootstrap bootstrap = new Bootstrap();
  34.         bootstrap.group(group).channel(NioSocketChannel.class)
  35.                 .remoteAddress(new InetSocketAddress(host, port))
  36.                 .handler(new ChannelInitializer<SocketChannel>() {
  37.                     @Override
  38.                     protected void initChannel(SocketChannel ch) throws Exception {
  39.                         ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
  40.                         ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
  41.                         ch.pipeline().addLast(new NettyMQConsumerHandler());
  42.                     }
  43.                 });
  44.         try {
  45.             // 发起同步连接
  46.             ChannelFuture sync = bootstrap.connect().sync();
  47.             DeliveryInfoEntity deliveryInfoEntity = new DeliveryInfoEntity(null, queueName,
  48.                     false);
  49.             // 发送数据给netty服务端
  50.             sync.channel().writeAndFlush(deliveryInfoEntity);
  51.             sync.channel().closeFuture().sync();
  52.         } catch (Exception e) {
  53.  
  54.         } finally {
  55.             group.shutdownGracefully();
  56.         }
  57.     }
  58.  
  59.  
  60. }

NettyMQServerHandler

netty服务Handler  mq服务Handler

  1. package com.mayikt.handler;
  2.  
  3. import com.mayikt.entity.DeliveryInfoEntity;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.SimpleChannelInboundHandler;
  6. import org.apache.commons.lang3.StringUtils;
  7.  
  8. import java.util.*;
  9.  
  10. /**
  11.  * netty服务Handler
  12.  */
  13. public class NettyMQServerHandler extends SimpleChannelInboundHandler<DeliveryInfoEntity> {
  14.     /**
  15.      * mq存放所有的队列  key:队列名称，value:队列数据内容
  16.      */
  17.     private static Map<String, Queue> queues = new HashMap<String, Queue>();
  18.  
  19.     /**
  20.      * 存放我们消费者连接（点对点） key:队列名称，value:管道
  21.      */
  22.     private static Map<String, ChannelHandlerContext> ctxs = new HashMap<String, ChannelHandlerContext>();
  23.  
  24.     /**
  25.      * Netty的服务器端接受 客户端消息  MQ服务器端
  26.      *
  27.      * @param ctx 管道
  28.      * @param dInfo 客户端消息封装的Entity
  29.      * @throws Exception
  30.      */
  31.     @Override
  32.     protected void channelRead0(ChannelHandlerContext ctx, DeliveryInfoEntity dInfo) throws Exception {
  33.         // 队列名称
  34.         String queueName = dInfo.getQueueName();
  35.         if (StringUtils.isEmpty(queueName)) {
  36.             return;
  37.         }
  38.         Boolean connType = dInfo.getConnType();
  39.         // 如果是为true的情况下 为生产者角色
  40.         if (connType) {
  41.             // 处理生产者角色（投递消息）
  42.             procucterService(queueName, dInfo.getMsg(), ctx);
  43.             return;
  44.         }
  45.         consumer(queueName, ctx);
  46.  
  47.     }
  48.  
  49.     private void procucterService(String queueName, String msg, ChannelHandlerContext channelHandlerContext) {
  50.         Queue queue = queues.get(queueName);
  51.         if (queue == null) {
  52.             // 如果队列不存在的情况下，就创建
  53.             queue = new LinkedList();
  54.             queues.put(queueName, queue);
  55.         }
  56.         // 将消息缓存到队列中
  57.         queue.offer(msg);
  58.  
  59.         // ack应答（消息是否投递成功）
  60.         channelHandlerContext.writeAndFlush("“ack应答”消息投递成功");
  61.  
  62.         // 主动将消息推送给消费者
  63.         ChannelHandlerContext ctx = ctxs.get(queueName);
  64.         if (ctx != null) {
  65.             ctx.writeAndFlush(queue.poll());
  66.         }
  67.  
  68.     }
  69.  
  70.     /**
  71.      * 消费和mq建立连接主动拉取消息
  72.      */
  73.     private void consumer(String queueName, ChannelHandlerContext ctx) {
  74.         Queue queue = queues.get(queueName);
  75.         if (queue == null) {
  76.             return;
  77.         }
  78.         // 获取队列中消息
  79.         Object poll = queue.poll();
  80.         ctx.writeAndFlush(poll);
  81.  
  82.         // 将消费者连接存放到集合中
  83.         ctxs.put(queueName, ctx);
  84.     }
  85.  
  86. }

NettyMQProducerHandler

生产者Handler

  1. package com.mayikt.handler;
  2.  
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.SimpleChannelInboundHandler;
  5.  
  6.  
  7. public class NettyMQProducerHandler extends SimpleChannelInboundHandler {
  8.  
  9.     protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object ack) throws Exception {
  10.         System.out.println(ack);
  11.        // 关闭连接
  12.         channelHandlerContext.close();
  13.     }
  14. }

NettyMQConsumerHandler

消费者Handler

  1. package com.mayikt.handler;
  2.  
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.SimpleChannelInboundHandler;
  5.  
  6. /**
  7.  * 消费者Handler
  8.  */
  9. public class NettyMQConsumerHandler extends SimpleChannelInboundHandler {
  10.     @Override
  11.     protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
  12.         System.out.println("消费者获取生产者消息:" + msg);
  13.     }
  14. }

DeliveryInfoEntity

传输协议的封装Entity - 传输协议如：amqp

  1. package com.mayikt.entity;
  2.  
  3. import lombok.Data;
  4.  
  5. import java.io.Serializable;
  6.  
  7. /**
  8.  * 传输协议封装Entity - 传输协议如：amqp
  9.  */
  10. @Data
  11. public class DeliveryInfoEntity implements Serializable {
  12.  
  13.     /**
  14.      * 发送消息内容
  15.      */
  16.     private String msg;
  17.  
  18.     /**
  19.      * 队列名称
  20.      */
  21.     private String queueName;
  22.  
  23.     /**
  24.      * true 生产者投递消息
  25.      * false 消费者获取消息
  26.      */
  27.     private Boolean connType;
  28.  
  29.     public DeliveryInfoEntity(String msg, String queueName, Boolean connType) {
  30.         this.msg = msg;
  31.         this.queueName = queueName;
  32.         this.connType = connType;
  33.     }
  34. }

标签：netty,mayikt,new,MQ,io,消息中间件,import,channel
From： https://www.cnblogs.com/LoveShare/p/17011362.html

相关文章

  • 7张图揭晓RocketMQ存储设计的精髓
    简介：RocketMQ作为一款基于磁盘存储的中间件，具有无限积压能力，并提供高吞吐、低延迟的服务能力，其最核心的部分必然是它优雅的存储设计。RocketMQ作为一款基于磁盘存储的中......
  • Linux-CentOS7安装RocketMQ
    下载地址https://archive.apache.org/dist/rocketmq/基础安装#解压unziprocketmq-all-4.7.0-bin-release.zip修改bin/runserver.shJAVA_OPT="${JAVA_OPT}-server......
  • 一分钟搞定Netty 三大组件，如果搞不定，再看3遍
    1.三大组件简介Channel与BufferJavaNIO系统的核心在于：通道(Channel)和缓冲区(Buffer)。通道表示打开到IO设备(例如：文件、套接字)的连接。若需要使用NIO系......
  • RocketMq基础
    各个mq比较ActiveMq和RabbitMq单机吞吐量是万级的，RocketMq和kafka是十万级。ActiveMq可能丢失数据，RocketMq可以做到零丢失。 RocketMq四个核心组成部分NameService:......
  • RabbitMQ从入门到精通-工作队列-Work Queues
         工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队......
  • Netty4底层用对象池和不用对象池实践优化
    随着JVM虚拟机和JIT即时编译技术的发展，对象的分配和回收是个非常轻量级的工作。但是对于缓冲区Buffer，情况却稍有不同，特别是对于堆外直接内存的分配和回收，是一件耗时的操作。......
  • netty的异常分析 IllegalReferenceCountException refCnt: 0, decrement: 1
    在websocket关闭时经常会抛出如下异常: IllegalReferenceCountExceptionrefCnt:0,decrement:1io.netty.util.IllegalReferenceCountException:refCnt:0,decrement:......
  • 高并发下Netty4底层bug导致直接内存溢出分析
    事故记录:10点游戏开服，迅速冲破2300+单区同时在线18点15分，运营反应玩家进不了，准备吃饭的人被抓回来排查故障发现，由于直接内存被占满，一直在FullGC,并且回收不掉，所以完全不......
  • 在Netty底层监控消息发送到Socket的时间
    1、调用writeAndFlush方法之后获取ChannelFuture；2、新增消息发送ChannelFutureListener，监听消息发送结果，如果消息写入网络Socket成功，则Netty会回调C......
  • Netty4 读写水位控制分析
    服务器上看看默认是多少ConfigurehighandlowwritewatermarksSetsane WRITE_BUFFER_HIGH_WATER_MARK andWRITE_BUFFER_LOW_WATER_MARKServerBootstrapbootstrap=......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99