This is the 9th day of my participation in Gwen Challenge
This article is participating in “Java Theme Month – Java Development in Action”, see the activity link for details
In the previous article, we introduced implementing an RPC framework based on Netty. In addition, Netty is also widely used in the work of instant messaging technology, one of the solutions, today we will take a look, based on Netty how to achieve a simple group chat system.
In the server startup code, create two EventLoopGroup thread groups. BossGroup is responsible for receiving client connections, and workerGroup is responsible for reading and writing network connections. After that, the ServerBootstrap server boot class is used to configure the entire Netty program, and all components are connected in series to start the program. The configuration of parameters will not be described here.
public class GroupChatServer {
private int port;
public GroupChatServer(int port) {
this.port = port;
}
public void run(a) {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decoder".new StringDecoder());
pipeline.addLast("encoder".new StringEncoder());
pipeline.addLast(newGroupChatServerHandler()); }}); System.out.println("NETTY SERVER IS READY");
ChannelFuture channelFuture = bootstrap.bind(port).sync();
channelFuture.channel().closeFuture().sync();
}catch (Exception e){
e.printStackTrace();
}finally{ bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); }}}Copy the codeThe core logic for server startup is as follows:
- For pipeline
pipeline - to
pipelineTo add decoders and encoders - Add service processor
handler
Specific business processor in the following definition, inheritance SimpleChannelInboundHandler class, processing the message received:
public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {
private static ChannelGroup channelGroup=newDefaultChannelGroup(GlobalEventExecutor.INSTANCE); . }Copy the codeStart by defining a ChannelGroup that manages all channels. GlobalEventExecutor. Is a global event INSTANCE said actuator, is a singleton pattern.
Here are a few core methods that need to be rewritten:
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
channelGroup.writeAndFlush("[client]"+channel.remoteAddress()+"Join the chat"+sdf.format(new Date())+"\n");
channelGroup.add(channel);
}
Copy the codeHandlerAdded indicates that a connection is created and is executed first. In this method, the current channel is added to the channelGroup. In addition, the chat information of the client is pushed to other online clients, which traverses all channels in the channelGroup and sends messages.
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
channelGroup.writeAndFlush("[client]"+channel.remoteAddress()+"Left \n");
System.out.println("channelGroup size:"+channelGroup.size());
}
Copy the codeDisconnecting triggers the handlerRemoved method. In this method, offline messages are pushed to customers who are currently online. Note that the current method is already executed when executed:
channelGroup.remove(channel);
Copy the codeThe remove method is executed automatically, so there is no need to call the remove method manually.
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println(ctx.channel().remoteAddress()+"Online");
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println(ctx.channel().remoteAddress()+"Offline");
}
Copy the codeThe channelActive and channelInactive methods indicate that a channel is active or inactive, and only up and down information is printed.
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
Channel channel = ctx.channel();
channelGroup.forEach(ch->{
if(channel! =ch){ ch.writeAndFlush("[customer]"+channel.remoteAddress()+"Sent a message :"+msg+"\n");
}else{display the message you sent ch.writeandFlush ("[himself] sent a message :"+msg+"\n"); }}); }Copy the codeChannelRead0 () is the read data method, which iterates through the channelGroup, sending different messages depending on the situation.
- If not the current one
channel, then it shows that other clients sent messages - If it’s current
channel“Displays that you sent the message
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
Copy the codeFinally, execute the exceptionCaught method to close the ChannelHandlerContext when an exception occurs.
The detailed code of the client is as follows:
public class GroupChatClient {
private final String host;
private final int port;
public GroupChatClient(String host, int port) {
this.host = host;
this.port = port;
}
public void run(a){
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
try{
Bootstrap bootstrap = new Bootstrap()
.group(eventLoopGroup)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decoder".new StringDecoder());
pipeline.addLast("encoder".new StringEncoder());
pipeline.addLast(newGroupChatClientHandler()); }}); System.out.println("NETTY CLIENT IS READY");
ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
Channel channel = channelFuture.channel();
System.out.println("--"+channel.localAddress()+"-");
Scanner scanner=new Scanner(System.in);
while (scanner.hasNext()){
String msg = scanner.nextLine();
channel.writeAndFlush(msg+"\r\n"); }}catch (Exception e){
e.printStackTrace();
}finally{ eventLoopGroup.shutdownGracefully(); }}}Copy the codeThe process of getting the pipe, adding the codec to the pipe, and setting up the business processor handler is essentially the same as on the server side. Significantly different from the server, the client needs to enter information to send to others, so it creates a scanner that receives input from the keyboard.
Client business processors, the same inheritance SimpleChannelInboundHandler:
public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { System.out.println(msg.trim()); }}Copy the codeCompared to the server side is very simple, just need to print the received message.
To test this process, start one server and three clients, and print the online information of the clients on the server:
Information printed when the server is started:
The client that logs in earlier will receive the online information of other clients forwarded by the server:
The send and receive message test will display the difference according to the sender:
When other clients go offline, the offline information of the current client is displayed:
The offline information displayed on the server:
At this point, the basic functions of a group chat system have been realized. Note that peer-to-peer chat is not supported here. If peer-to-peer chat is needed, then ChannelGroup can not be used. Instead, HashMap can be used to cache the information of each Channel to achieve targeted message sending.
The last
If you think it is helpful, you can like it and forward it. Thank you very much
Wechat search: code agricultural ginseng, to add a friend, like the friends can also oh ~
Public account background reply “interview”, “map”, “structure”, “actual combat”, get free information oh ~