IM – 单聊
1. 前言
本节,我们主要来实现单聊功能,分别登陆两个账 号,然后可以互相发送消息。具体效果图如下所示:
客户端 1 登陆效果:
客户端 2 登陆效果:
客户端 1 发送消息效果图:
客户端 2 接受消息效果图:
2. 实现流程
以上效果是一个基于 console 控制台的聊天效果,根据上节的思路设计,我们也知道主要核心是服务端保存一份关系映射,通过接受人 ID 找到对应的通道进行消息发送。但是,我们实现具体的功能,需要大体上做一个核心步骤的拆解,具体如下所示:
第一步: 编码和解码的实现。
两个核心功能点,第一是序列化和反序列化,第二通讯协议实现。
客户端和服务端之间的数据通讯,我们是基于实体对象去交互,这样数据格式更加的方便,因此,然而对于实体对象对象的序列化和反序列化推荐使用 Fastjson 框架去实现,而不是前面章节所使用的对象流。同时为了更加规范的管理不同业务的实体,我们需要定义一个实体基类,所有的业务实体都继承它(下面会详细讲解)。
第二步: 登录和消息发送两个业务点的实现。
登录主要是为了让用户 ID 和通道进行绑定,在登录成功之后给 Channel 通过 attr () 方法绑定用户 ID,主要目的有两个:
- 在发送消息时,可以通过 Channel 获取用户 ID,发送给服务端,让另外一个客户端接收消息时,知道是谁发送过来的消息;
- 在 Channel 断开的时候,服务端可以监听到 Channel,并且获取 Channel 的属性,从而删除对应的映射关系。
业务处理,用户登录和消息发送是两个不同的业务点,一般来说需要定义多个 Handler 来分别处理,但是这里为了减少 Handler 的数量,统一一个 Handler 处理。
第三步: 映射关系的实现。
前面也解析过了,服务端需要保存一份映射关系,只需要使用一个 Map 进行存储即可,Map<Integer,Channel>
,key 是用户 ID,value 是 Channel,也就是连接通道。
3. 功能实现
3.1 定义实体
实体的设计,主要从两个方面进行入手考虑,第一: 通讯协议的规则,分别是协议标识符、业务指令、数据长度、数据四个部分;第二: 不同业务对应不同的字段属性。具体如下图所示:
基础实体
主要定义 tag 字段,标识协议的标识符,code () 抽象方法,主要表示的是业务指令,不同的业务对应不同的指令。
@Data
public abstract class BaseBean implements Serializable {
private Integer tag=1;//固定值,标识的是一个协议类型,不同协议对应不同的值
public abstract Byte code();//业务指令抽象方法
}
登录请求实体
@Data
public class LoginReqBean extends BaseBean implements Serializable {
private Integer userid;//用户ID
private String username;//用户名称
public Byte code() {
return 1;//业务指令
}
}
登录响应实体
@Data
public class LoginResBean extends BaseBean implements Serializable {
private Integer status;//响应状态,0登录成功,1登录失败
private String msg;//响应信息
private Integer userid;//用户ID
public Byte code() {
return 2;//业务指令
}
}
消息发送实体
public class MsgReqBean extends BaseBean implements Serializable {
private Integer fromuserid;//发送人ID
private Integer touserid;//接受人ID
private String msg;//发送消息
public Byte code() {
return 3;//业务指令
}
}
消息响应响应
public class MsgResBean extends BaseBean implements Serializable {
private Integer status;//响应状态,0发送成功,1发送失败
private String msg;//响应信息
public Byte code() {
return 4;//业务指令
}
}
消息接受实体
public class MsgRecBean extends BaseBean implements Serializable {
private Integer fromuserid;//发送人ID
private String msg;//消息
public Byte code() {
return 5;//业务指令
}
}
3.2 编码和解码实现
3.2.1 依赖坐标
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.47</version>
</dependency>
3.2.2 编码实现
public class MyEncoder extends MessageToByteEncoder<BaseBean> {
protected void encode(
ChannelHandlerContext channelHandlerContext,
BaseBean baseBean,
ByteBuf byteBuf) throws Exception {
//1.把实体序列化成字节数字
byte[] bytes= JSON.toJSONBytes(baseBean);
//2.根据协议组装数据
byteBuf.writeInt(baseBean.getTag());//标识(4个字节)
byteBuf.writeByte(baseBean.code());//指令(1个字节)
byteBuf.writeInt(bytes.length);//长度(4个字节)
byteBuf.writeBytes(bytes);//
}
}
3.2.3 解码实现
public class MyDecoder extends ByteToMessageDecoder {
protected void decode(
ChannelHandlerContext channelHandlerContext,
ByteBuf byteBuf,
List<Object> list) throws Exception {
//1.根据协议取出数据
int tag=byteBuf.readInt();//标识符
byte code=byteBuf.readByte();//获取指令
int len=byteBuf.readInt();//获取数据长度
byte[] bytes=new byte[len];
byteBuf.readBytes(bytes);
//2.根据code获取类型
Class<? extends BaseBean> c= MapUtils.getBean(code);
//3.反序列化
BaseBean baseBean=JSON.parseObject(bytes,c);
list.add(baseBean);
}
}
3.2.4 指令和实体关系
为什么需要这么一个工具类呢?指令表示的是业务类型,不同的业务对应不同的实体,那么解码的时候,怎么知道反序列化成什么样的实体呢?思路是获取到的指令,再根据指令找到对应的实体即可。
public class MapUtils {
//1.自定义指令
private static Byte codeLoginReq=1;
private static Byte codeLoginRes=2;
private static Byte codeMsgReq=3;
private static Byte codeMsgRes=4;
private static Byte codeMsgRec=5;
//2.自定义一个Map,专门管理指令和实体的关系
private static Map<Byte, Class<? extends BaseBean>> map=new HashMap<Byte,Class<? extends BaseBean>>();
//3.初始化
static {
map.put(codeLoginReq, LoginReqBean.class);
map.put(codeLoginRes, LoginResBean.class);
map.put(codeMsgReq, MsgReqBean.class);
map.put(codeMsgRes, MsgResBean.class);
map.put(codeMsgRec, MsgRecBean.class);
}
//4.根据指令获取对应的实体
public static Class<? extends BaseBean> getBean(Byte code){
try{
return map.get(code);
}catch (Exception e){
throw new RuntimeException(e.getMessage());
}
}
}
3.3 客户端代码实现
3.3.1 Pipeline 管理链表
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
//1.拆包器
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,5,4));
//2.自定义解码器
ch.pipeline().addLast(new MyDecoder());
//3.自定义业务
ch.pipeline().addLast(new ClientChatHandler());
//4.自定义编码器
ch.pipeline().addLast(new MyEncoder());
}
});
3.3.2 业务 Handler
所有的业务处理在同一个 Handler 里面进行处理,通过判断实体类型来区分不同的业务处理。
public class ClientChatHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
//通道就绪时,发起登录请求
login(ctx.channel());
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//根据msg做类型判断,不同的业务做不同的处理
if(msg instanceof LoginResBean){
//1.登录结果响应
LoginResBean res=(LoginResBean) msg;
System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>登录响应:"+res.getMsg());
if(res.getStatus()==0){
//1.登录成功,则给通道绑定属性
ctx.channel().attr(AttributeKey.valueOf("userid")).set(res.getUserid());
//2.调用发送消息方法
sendMsg(ctx.channel());
}else{
//1.登录失败,调用登录方法
login(ctx.channel());
}
}else if(msg instanceof MsgResBean){
//1.发送消息结果响应
MsgResBean res=(MsgResBean)msg;
System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>发送响应:"+res.getMsg());
}else if(msg instanceof MsgRecBean){
//2.接受消息
MsgRecBean res=(MsgRecBean)msg;
System.out.println("fromuserid="+res.getFromuserid()+",msg="+res.getMsg());
}
}
//登录方法
private void login(Channel channel){
Scanner scanner=new Scanner(System.in);
System.out.println(">>用户ID:");
Integer userid=scanner.nextInt();
System.out.println(">>用户名称:");
String username=scanner.next();
LoginReqBean bean=new LoginReqBean();
bean.setUserid(userid);
bean.setUsername(username);
channel.writeAndFlush(bean);
}
//发送消息方法
private void sendMsg(final Channel channel){
final Scanner scanner=new Scanner(System.in);
new Thread(new Runnable() {
public void run() {
while (true){
System.out.println(">>接收人ID:");
Integer touserid=scanner.nextInt();
System.out.println(">>聊天内容:");
String msg=scanner.next();
MsgReqBean bean=new MsgReqBean();
//从通道属性获取发送人ID
Integer fromuserid=(Integer) channel.attr(
AttributeKey.valueOf("userid")
).get();
//发送人ID
bean.setFromuserid(fromuserid);
//接受人ID
bean.setTouserid(touserid);
//发送消息
bean.setMsg(msg);
channel.writeAndFlush(bean);
}
}
}).start();
}
}
3.4 服务端代码实现
3.4.1 Pipeline 管理链表
.childHandler(new ChannelInitializer<NioSocketChannel>() {
protected void initChannel(NioSocketChannel ch) {
//1.拆包器
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,5,4));
//2.自定义解码器
ch.pipeline().addLast(new MyDecoder());
//3.业务Handler
ch.pipeline().addLast(new ServerChatHandler());
//4.自定义编码器
ch.pipeline().addLast(new MyEncoder());
}
});
3.4.2 业务 Handler
public class ServerChatHandler extends ChannelInboundHandlerAdapter{
//1.定义一个Map(key是用户ID,value是连接通道)
private static Map<Integer, Channel> map=new HashMap<Integer, Channel>();
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if(msg instanceof LoginReqBean){
//1.登录请求
login((LoginReqBean) msg,ctx.channel());
}else if(msg instanceof MsgReqBean){
//2.发送消息请求
sendMsg((MsgReqBean)msg,ctx.channel());
}
}
//登录处理方法
private void login(LoginReqBean bean, Channel channel){
LoginResBean res=new LoginResBean();
//从map里面根据用户ID获取连接通道
Channel c=map.get(bean.getUserid());
if(c==null){
//通道为空,证明该用户没有在线
//1.添加到map
map.put(bean.getUserid(),channel);
//2.给通道赋值
channel.attr(AttributeKey.valueOf("userid")).set(bean.getUserid());
//3.响应
res.setStatus(0);
res.setMsg("登录成功");
res.setUserid(bean.getUserid());
channel.writeAndFlush(res);
}else{
//通道不为空,证明该用户已经在线了
res.setStatus(1);
res.setMsg("该账户目前在线");
channel.writeAndFlush(res);
}
}
//消息发送处理方法
private void sendMsg(MsgReqBean bean,Channel channel){
Integer touserid=bean.getTouserid();
Channel c=map.get(touserid);
if(c==null){
MsgResBean res=new MsgResBean();
res.setStatus(1);
res.setMsg(touserid+",不在线");
channel.writeAndFlush(res);
}else{
MsgRecBean res=new MsgRecBean();
res.setFromuserid(bean.getFromuserid());
res.setMsg(bean.getMsg());
c.writeAndFlush(res);
}
}
}
4. 小结
本节主要实现了单聊功能,实现思路相对还是稍微有点小复杂,大家主要核心掌握以下几点思路:
- 不同业务设置不同的实体和指令,需要把指令和实体的关系管理起来,方便反序列化的时候,可以根据指令来反序列化到具体实体;
- 需要把用户 ID 和通道的关系管理起来,并且灵活运用 Channel 的 attr () 方法,该方法可以绑定属性值,非常的有用。
最新评论
Spring Cloud Alibaba 微服务架构实战 https://pan.baidu.com/s/1jF5voFRoeF0lYAzAPBWSbw?pwd=chqk
命令: nload
真是个良心站点哇,大公无私,爱了爱了
还可以直接搞一张映射表,存 uid | time | source_index, 第一次直接查对应的 time 选出前100, 第二次直接用 CompleteFuture 去分别用 source_in
干得漂亮,多个朋友堵条路
2021.2.2版本的不适用吧
现在还可以用么
激活码有用,感谢分享