本文由作者“大白菜”分享,个人博客 cmsblogs.cn,即时通讯收录时有较多修订和改动。注意:本系列是给IM初学者的文章,IM老油条们还望海涵,勿喷!
cover-opti.png (11.04 KB, 下载次数: 1394)
下载附件 保存到相册
2 年前 上传
x.jpg (17.09 KB, 下载次数: 1376)
1.png (13.4 KB, 下载次数: 1317)
2.png (13.18 KB, 下载次数: 1391)
3.png (6.91 KB, 下载次数: 1417)
4.png (6.9 KB, 下载次数: 1596)
5.png (29.84 KB, 下载次数: 1468)
@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;//业务指令 } }
<dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.47</version> </dependency>
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);// } }
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); } }
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()); } } }
.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()); } });
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(); } }
.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()); } });
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); } } }
来源:即时通讯网 - 即时通讯开发者社区!
轻量级开源移动端即时通讯框架。
快速入门 / 性能 / 指南 / 提问
轻量级Web端即时通讯框架。
详细介绍 / 精编源码 / 手册教程
移动端实时音视频框架。
详细介绍 / 性能测试 / 安装体验
基于MobileIMSDK的移动IM系统。
详细介绍 / 产品截图 / 安装体验
一套产品级Web端IM系统。
详细介绍 / 产品截图 / 演示视频
引用此评论
引用:老衲 发表于 2022-07-15 11:28 这里可以不用LengthFieldBasedFrameDecoder,直接使用ByteToMessageDecoder读取协议中的长度判断消息长度解 ...
引用:准社畜老六 发表于 2022-07-20 14:29 作者你好,想问一下如果发送文本和二进制文件的话要分开用不同的handler处理,还是用直接全部用二进制处理的 ...
引用:JackJiang 发表于 2022-07-20 16:24 直接二进制,然后在内容自已加类型字段,这样更灵活
引用:准社畜老六 发表于 2022-07-20 19:23 好滴,谢谢
精华主题数超过100个。
连续任职达2年以上的合格正式版主
为论区做出突出贡献的开发者、版主等。
Copyright © 2014-2024 即时通讯网 - 即时通讯开发者社区 / 版本 V4.4
苏州网际时代信息科技有限公司 (苏ICP备16005070号-1)
Processed in 0.171875 second(s), 44 queries , Gzip On.