1. 服务端改造

1.1 修改 Message类

package com.hahashou.netty.server.config;import com.alibaba.fastjson.JSON;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.CharsetUtil;
import lombok.Data;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
@Data
public class Message {/** 发送者用户code */private String userCode;/** 接收者用户code */private String friendUserCode;/** 连接时专用 */private String channelId;/** 文字 */private String text;public static ByteBuf transfer(Message message) {return Unpooled.copiedBuffer(JSON.toJSONString(message), CharsetUtil.UTF_8);}
}

1.2 修改 NettyServerHandler类

package com.hahashou.netty.server.config;import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
@Component
@ChannelHandler.Sharable
@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {/** key: 用户code; value: channelId */public static Map<String, String> USER_CHANNEL = new ConcurrentHashMap<>(32);/** key: channelId; value: Channel */public static Map<String, Channel> CHANNEL = new ConcurrentHashMap<>(32);@Overridepublic void channelActive(ChannelHandlerContext ctx) {Channel channel = ctx.channel();String channelId = channel.id().asLongText();log.info("有客户端连接, channelId : " + channelId);CHANNEL.put(channelId, channel);Message message = new Message();message.setChannelId(channelId);channel.writeAndFlush(Message.transfer(message));}@Overridepublic void channelInactive(ChannelHandlerContext ctx) {log.info("有客户端断开连接, channelId : {}", ctx.channel().id().asLongText());CHANNEL.remove(ctx.channel().id().asLongText());}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {if (msg != null) {Message message = JSON.parseObject(msg.toString(), Message.class);String userCode = message.getUserCode(),channelId = message.getChannelId(),text = message.getText();if (StringUtils.hasText(userCode) && StringUtils.hasText(channelId)) {log.info("客户端 " + userCode + " 连接");USER_CHANNEL.put(userCode, channelId);} else if (StringUtils.hasText(text)) {String friendUserCode = message.getFriendUserCode();if (StringUtils.hasText(friendUserCode)) {// 向某客户端发送String queryChannelId = USER_CHANNEL.get(friendUserCode);if (StringUtils.hasText(queryChannelId)) {Channel channel = CHANNEL.get(queryChannelId);if (channel == null) {log.error(friendUserCode + "已下线, 存储消息 :{}", text);return;}channel.writeAndFlush(Message.transfer(message));} else {log.error("无此用户信息");}} else {// 向服务端发送时, 返回当前时间message.setUserCode("ADMIN");message.setText(LocalDateTime.now().toString());ctx.channel().writeAndFlush(Message.transfer(message));}}}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {log.info("有客户端发生异常, channelId : {}", ctx.channel().id().asLongText());}
}

1.3 修改 ServerController类

package com.hahashou.netty.server.controller;import com.hahashou.netty.server.config.Message;
import com.hahashou.netty.server.config.NettyServerHandler;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
@RestController
@RequestMapping("/server")
@Slf4j
public class ServerController {@PostMapping("/send")public String send(@RequestBody Message dto) {String friendUserCode = dto.getFriendUserCode();if (StringUtils.hasText(friendUserCode)) {String channelId = NettyServerHandler.USER_CHANNEL.get(friendUserCode);if (StringUtils.hasText(channelId)) {Channel channel = NettyServerHandler.CHANNEL.get(channelId);if (channel == null) {return "该客户端已下线, 需要存储相关信息";}channel.writeAndFlush(Message.transfer(dto));} else {log.error("无此用户信息");}} else {// 全体广播NettyServerHandler.USER_CHANNEL.forEach((userCode, channelId) -> {Channel channel = NettyServerHandler.CHANNEL.get(channelId);if (channel == null) {log.error(userCode + "已下线, 需要存储相关信息");return;}channel.writeAndFlush(Message.transfer(dto));});}return "success";}
}

1.4 启动服务端

2. 客户端改造

2.1 复制服务端的 Message类

2.2 修改 NettyClientHandler类

package com.hahashou.netty.client.config;import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
@Component
@ChannelHandler.Sharable
@Slf4j
public class NettyClientHandler extends ChannelInboundHandlerAdapter {@Value("${userCode}")private String userCode;public static Channel CHANNEL;@Overridepublic void channelActive(ChannelHandlerContext ctx) {CHANNEL = ctx.channel();log.info("客户端 " + userCode + " 上线");}@Overridepublic void channelInactive(ChannelHandlerContext ctx) {CHANNEL = null;}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {if (msg != null) {Message message = JSON.parseObject(msg.toString(), Message.class);String channelId = message.getChannelId(),text = message.getText();if (StringUtils.hasText(channelId)) {Channel channel = ctx.channel();message.setUserCode(userCode);channel.writeAndFlush(Message.transfer(message));} else if (StringUtils.hasText(text)) {log.info("收到" + message.getUserCode() + "消息: {}", text);}}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {CHANNEL = null;}
}

2.3 修改 ClientController类

package com.hahashou.netty.client.controller;import com.hahashou.netty.client.config.Message;
import com.hahashou.netty.client.config.NettyClientHandler;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;/*** @description:* @author: 哼唧兽* @date: 9999/9/21**/
@RestController
@RequestMapping("/client")
@Slf4j
public class ClientController {@PostMapping("/send")public String send(@RequestBody Message dto) {Channel channel = NettyClientHandler.CHANNEL;if (channel == null) {return "服务端已下线";}channel.writeAndFlush(Message.transfer(dto));return "success";}
}

3. 通过修改配置信息, 生成2个客户端

3.1 A客户端 application.yml, 打包jar

server:port: 32001logging:level:com.hahashou.netty: infouserCode: Aa

3.2 B客户端 application.yml, 打包jar

server:port: 32002logging:level:com.hahashou.netty: infouserCode: Bb

3.3 启动A、B客户端

4. 测试

4.1 A与B互发消息

根据上述示例, 以及下述图片, 自行补充相关信息(注意服务的切换)

4.2 A客户端向服务端发送消息

4.3 服务端向A单独通知以及全体广播

A客户端

B客户端