主程序启动模版

package net.jlxxw.wechat.temp;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import net.jlxxw.wechat.component.netty.WeChatCoreComponent;
import net.jlxxw.wechat.util.LoggerUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/*** @author chunyang.leng* @date 2022-04-27 9:35 AM*/
public class Main {private static final ServerBootstrap BOOTSTRAP = new ServerBootstrap();private static final Logger logger = LoggerFactory.getLogger(WeChatCoreComponent.class);public static void main(String[] args) {LoggerUtils.info(logger, "初始化 netty 组件");//new 一个主线程组EventLoopGroup bossGroup = new NioEventLoopGroup(1);//new 一个工作线程组EventLoopGroup workGroup = new NioEventLoopGroup(200);BOOTSTRAP.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {// todo 编解码器 略socketChannel.pipeline().addLast("channel_manager",new AllChannelManager());// 自定义处理handlersocketChannel.pipeline().addLast("chatChannel", new ChatChannel());LoggerUtils.debug(logger, "初始化 netty 微信协议处理器 成功");}}).localAddress(9999)//设置队列大小.option(ChannelOption.SO_BACKLOG, 200)// 不保持长链接.childOption(ChannelOption.SO_KEEPALIVE, false);//绑定端口,开始接收进来的连接try {ChannelFuture future = BOOTSTRAP.bind(9999).sync();LoggerUtils.info(logger, "netty服务启动，开始监听端口: {}", 9999);future.channel().closeFuture().sync();} catch (InterruptedException e) {LoggerUtils.error(logger, "netty服务启动失败！！！", e);System.exit(0);} finally {//关闭主线程组bossGroup.shutdownGracefully();//关闭工作线程组workGroup.shutdownGracefully();}}}

业务处理channel

package net.jlxxw.wechat.temp;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import net.jlxxw.wechat.util.LoggerUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;import java.nio.charset.StandardCharsets;/*** netty回调处理接口** @author chunyang.leng* @date 2021/1/25 9:46 上午*/
@Component
@ChannelHandler.Sharable
public class ChatChannel extends SimpleChannelInboundHandler<ByteBuf> {private static final Logger logger = LoggerFactory.getLogger(ChatChannel.class);@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {LoggerUtils.debug(logger, "netty 开始处理");// 请求体数据转byte数组byte[] reqContent = new byte[byteBuf.readableBytes()];// 缓存数据加载至byte数组中byteBuf.readBytes(reqContent);final String requestData = new String(reqContent, StandardCharsets.UTF_8);logger.info("接收到请求数据:{}",requestData);AllChannelManager.sendAll(requestData);}/*** 异常信息记录* @param ctx* @param cause* @throws Exception*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {// 记录异常日志LoggerUtils.error(logger,"netty-thread 发生未知异常",cause);// 关闭异常连接ctx.close();}
}

channel管理器

package net.jlxxw.wechat.temp;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.ImmediateEventExecutor;/*** @author chunyang.leng* @date 2022-04-27 9:41 AM*/
public class AllChannelManager extends ChannelDuplexHandler {/*** ChannelGroup用来保存所有已经连接的Channel*/private final static ChannelGroup CHANNEL_GROUP = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {CHANNEL_GROUP.add(ctx.channel());super.channelActive(ctx);}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {CHANNEL_GROUP.remove(ctx.channel());super.channelInactive(ctx);}/*** 发送给所有人* @param message*/public static void sendAll(String message){CHANNEL_GROUP.parallelStream().filter(Channel::isActive).forEach(channel -> {ByteBuf byteBuf = Unpooled.directBuffer(message.length());byteBuf.writeCharSequence(message, CharsetUtil.UTF_8);channel.writeAndFlush(byteBuf);});}
}

netty 实现消息群发相关推荐

netty实现消息群发
netty是什么我所理解的netty,是一个比较底层的网络编程的框架,它和tomcat的区别是什么呢?就是tomcat是一个已经封装好的容器,你可以直接使用,而netty是可以写出像tomcat这种 ...
java 微信群发多图文_[Java教程]httpClient实现微信公众号消息群发
[Java教程]httpClient实现微信公众号消息群发 0 2016-09-21 20:00:10 1.实现功能向关注了微信公众号的微信用户群发消息.(可以是所有的用户,也可以是提供了微信ope ...
微信公众号开发 [03] 结合UEditor实现图文消息群发功能
0.写在前面的话如何实现微信平台后台管理中的,图文消息发送功能? 大概的过程如下: 通过类似表单的形式,将文章各部分内容提交到后台,封装成一个实体类,并持久化到数据库中需要推送的时候,将不同的文章 ...
公众平台模板消息所在行业_第三方工具微信公众号模板消息群发如何操作？
当下,公众平台模板消息功能仅支持添加模板,修改所在行业,如果想要群发模板消息,可以自己根据公众平台的接口编程实现,也可通过微号帮平台的模板消息群发功能实现,均可以让微信公众号群发模板消息,模板消息即按 ...
公众平台模板消息所在行业_微信公众号模板消息群发第三方平台可以免费试用吗？...
使用公众号后台模板消息功能,只能添加模板和修改所在行业,如果想要群发模板消息,可以通过公众号后台的接口自己编程实现,也可使用第三方平台微号帮功能模板消息群发,都可以实现微信公众号模板消息群发,突破公众 ...
SpringBoot 集成 WebSocket 实现消息群发推送
一. 什么是 WebSocket WebSocket 是一种全新的协议.它将 TCP 的 Socket(套接字)应用在了web page上,从而使通信双方建立起一个保持在活动状态的连接通道,并且属于全 ...
公众号客服消息超过48小时_免费模板消息群发的方法，在这里！
不知道大家会不会因为服务号4次推送机会用完,又遇到老板施压,让发布重要消息,而苦恼? 经过我苦心搜索,找到了一个解决方法!原理就是,利用公众号模板消息,给粉丝推送. 作为运营同学,大多是不会技术的,所 ...
使用模板消息及利用模板消息群发的说明
下周就是国庆节,不知道你公众号的群发次数用完了吗?如果用完了,还可以使用模板消息群发对全部粉丝推送,仅限服务号使用.订阅号依然每天只能用群发1次,开发公众号高级群发接口,不会开发的人员可以直接使用微号 ...
java BIO tcp服务端向客户端消息群发代码教程实战
前言项目需要和第三方厂商的服务需要用TCP协议通讯,考虑到彼此双方可能都会有断网重连.宕机重启的情况,需要保证发生上述情况后,服务之间能够自动实现重新通信.研究测试之后整理如下代码实现.因为发现客 ...

netty 实现消息群发

主程序启动模版

业务处理channel

channel管理器

netty 实现消息群发相关推荐

最新文章

热门文章