基于Netty+MQTT的高性能推送服务框架。支持普通Socket、MQTT、MQTT web socket协议。非常方便接入上层业务实现推送业务。
这是一个基于Netty4.x + MQTT实现的Push推送基础框架。相比于原生Netty, socket-mqtt:
# 允许回收TCP连接 net.ipv4.tcp_tw_reuse = 1 net.ipv4.tcp_tw_recycle = 1TCP 缓冲区内存
net.ipv4.tcp_rmem = 4096 87380 8388608 net.ipv4.tcp_wmem = 4096 87380 8388608 net.ipv4.tcp_mem = 94500000 915000000 927000000
ulimits 优化
fs.file-max = 1065353 kernel.pid_max = 65536
单Broker8核16G,支持44万连接;1万客户端 单消息1024B 下行tps: 16万+; 4000客户端 Publish 单消息1024B 上行tps: 17万+,千兆网卡流量基本打满。 备注:Mqtt Server启动内存只分配了5G,如果分配到10G,理论上可以支持百万连接。还有,测试开启了心跳上报。
1万Clients订阅的消息下行能力 | 对应下行负载情况 |
---|---|
![]() |
![]() |
4000Clients订阅消息上行能力 | 对应上行负载情况 |
---|---|
![]() |
![]() |
查看连接数(telnet 10.43.204.61 8001; get status) | 查看连接数(ss -l) |
---|---|
![]() |
![]() |
各种测试类的源码在src/test/java/com/yb/socket包路径下: 包括: * 普通socket Server/Client * MQTT socket Server/Client * 带注册中心的普通socket/MQTT socket * 基于内存数据库的模拟订阅推送
Server server = new Server(); // 设置Broker端口 server.setPort(8000); // 设置启动信息统计。默认true server.setOpenCount(true); // 设置启用心跳功能。默认true server.setCheckHeartbeat(true); // 设置启动服务状态,默认端口8001。通过telnet server_ip 8001; get status查看服务信息 server.setOpenStatus(true); // 服务状态端口。默认8001 server.setStatusPort(8001); // 设置服务名称 server.setServiceName("Demo"); // 设置工作线程数量。默认CPU个数+1 server.setWorkerCount(64); // 是否开户业务处理线程池。默认false server.setOpenExecutor(true); // 设置tcp no delay。默认true server.setTcpNoDelay(true); // 是否启用keepAlive。默认true server.setKeepAlive(true); // 自定义监听器,可处理相关事件 server.addEventListener(new EchoMessageEventListener()); // 设置Broker启动协议。SocketType.MQTT - MQTT协议; SocketType.NORMAL - 普通Socket协议;SocketType.MQTT_WS - MQTT web socket协议; server.setSocketType(SocketType.MQTT); // 绑定端口启动服务 server.bind();
Server server = new Server(); server.setPort(8000); server.addEventListener(new EchoMessageEventListener()); server.setSocketType(SocketType.MQTT_WS); server.bind();//模拟推送 String message = "this is a web socket message!"; MqttRequest mqttRequest = new MqttRequest((message.getBytes())); while (true) { if (server.getChannels().size() > 0) { logger.info("模拟推送消息"); for (WrappedChannel channel : server.getChannels().values()) { server.send(channel, "yb/notice/", mqttRequest); } } Thread.sleep(1000L); }
可用在线mqtt测试:http://www.tongxinmao.com/txm/webmqtt.php Topic Payload Time QoS yb/notice/ this is a web socket message! 2019-2-27 16:54:54 0
Server server = new Server(); server.setPort(8000); server.addEventListener(new JsonEchoMessageEventListener()); server.addChannelHandler("decoder", new JsonDecoder()); server.addChannelHandler("encoder", new JsonEncoder()); server.bind();//模拟推送 JSONObject message = new JSONObject(); message.put("action", "echo"); message.put("message", "this is a normal socket message!");
Request request = new Request(); request.setSequence(0); request.setMessage(message); while (true) { if (server.getChannels().size() > 0) { logger.info("模拟推送消息"); for (WrappedChannel channel : server.getChannels().values()) { channel.send(request); Thread.sleep(5000L); } } }
Client client = new Client(); client.setIp("127.0.0.1"); client.setPort(8000); client.setConnectTimeout(10000); client.addChannelHandler("decoder", new JsonDecoder()); client.addChannelHandler("encoder", new JsonEncoder()); client.connect();for (int i = 0; i < 2; i++) { JSONObject message = new JSONObject(); message.put("action", "echo"); message.put("message", "hello world!");
Request request = new Request(); request.setSequence(i); request.setMessage(message); Response response = (Response) client.sendWithSync(request, 3000); logger.info("成功接收到同步的返回: '{}'.", response);
}
client.shutdown();
Server server = new Server(); server.setPort(9000); server.setCheckHeartbeat(false); server.addChannelHandler("decoder", new JsonDecoder()); server.addChannelHandler("encoder", new JsonEncoder()); server.addEventListener(new com.yb.socket.center.CenterMockMessageEventListener()); server.bind();
Server server = new Server(); server.setPort(8000); server.setCheckHeartbeat(false); server.setCenterAddr("127.0.0.1:9000,127.0.0.1:9010"); server.addEventListener(new JsonEchoMessageEventListener()); server.bind();
Client client = new Client(); client.setCheckHeartbeat(false); client.setCenterAddr("127.0.0.1:9000,127.0.0.1:9010"); client.addChannelHandler("decoder", new JsonDecoder()); client.addChannelHandler("encoder", new JsonEncoder()); client.connect();JSONObject message = new JSONObject(); message.put("action", "echo"); message.put("message", "hello");
for (int i = 0; i < 5; i++) { Request request = new Request(); request.setSequence(i); request.setMessage(message); client.send(request); Thread.sleep(5000L); }