曼巴out ·IM - 單聊
1. 前言
本節(jié)，我們主要來(lái)實(shí)現(xiàn)單聊功能，分別登陸兩個(gè)賬 號(hào)，然后可以互相發(fā)送消息。具體效果圖如下所示：
客戶端 1 登陸效果：
客戶端 2 登陸效果：
客戶端 1 發(fā)送消息效果圖：
客戶端 2 接受消息效果圖：
2. 實(shí)現(xiàn)流程
以上效果是一個(gè)基于 console 控制臺(tái)的聊天效果，根據(jù)上節(jié)的思路設(shè)計(jì)，我們也知道主要核心是服務(wù)端保存一份關(guān)系映射，通過(guò)接受人 ID 找到對(duì)應(yīng)的通道進(jìn)行消息發(fā)送。但是，我們實(shí)現(xiàn)具體的功能，需要大體上做一個(gè)核心步驟的拆解，具體如下所示：
第一步： 編碼和解碼的實(shí)現(xiàn)。
兩個(gè)核心功能點(diǎn)，第一是序列化和反序列化，第二通訊協(xié)議實(shí)現(xiàn)。
客戶端和服務(wù)端之間的數(shù)據(jù)通訊，我們是基于實(shí)體對(duì)象去交互，這樣數(shù)據(jù)格式更加的方便，因此，然而對(duì)于實(shí)體對(duì)象對(duì)象的序列化和反序列化推薦使用 Fastjson 框架去實(shí)現(xiàn)，而不是前面章節(jié)所使用的對(duì)象流。同時(shí)為了更加規(guī)范的管理不同業(yè)務(wù)的實(shí)體，我們需要定義一個(gè)實(shí)體基類，所有的業(yè)務(wù)實(shí)體都繼承它（下面會(huì)詳細(xì)講解）。
第二步： 登錄和消息發(fā)送兩個(gè)業(yè)務(wù)點(diǎn)的實(shí)現(xiàn)。
登錄主要是為了讓用戶 ID 和通道進(jìn)行綁定，在登錄成功之后給 Channel 通過(guò) attr () 方法綁定用戶 ID，主要目的有兩個(gè)：
- 在發(fā)送消息時(shí)，可以通過(guò) Channel 獲取用戶 ID，發(fā)送給服務(wù)端，讓另外一個(gè)客戶端接收消息時(shí)，知道是誰(shuí)發(fā)送過(guò)來(lái)的消息；
- 在 Channel 斷開的時(shí)候，服務(wù)端可以監(jiān)聽到 Channel，并且獲取 Channel 的屬性，從而刪除對(duì)應(yīng)的映射關(guān)系。
業(yè)務(wù)處理，用戶登錄和消息發(fā)送是兩個(gè)不同的業(yè)務(wù)點(diǎn)，一般來(lái)說(shuō)需要定義多個(gè) Handler 來(lái)分別處理，但是這里為了減少 Handler 的數(shù)量，統(tǒng)一一個(gè) Handler 處理。
第三步： 映射關(guān)系的實(shí)現(xiàn)。
前面也解析過(guò)了，服務(wù)端需要保存一份映射關(guān)系，只需要使用一個(gè) Map 進(jìn)行存儲(chǔ)即可，Map<Integer,Channel>，key 是用戶 ID，value 是 Channel，也就是連接通道。
3. 功能實(shí)現(xiàn)
3.1 定義實(shí)體
實(shí)體的設(shè)計(jì)，主要從兩個(gè)方面進(jìn)行入手考慮，第一： 通訊協(xié)議的規(guī)則，分別是協(xié)議標(biāo)識(shí)符、業(yè)務(wù)指令、數(shù)據(jù)長(zhǎng)度、數(shù)據(jù)四個(gè)部分；第二： 不同業(yè)務(wù)對(duì)應(yīng)不同的字段屬性。具體如下圖所示：
基礎(chǔ)實(shí)體
主要定義 tag 字段，標(biāo)識(shí)協(xié)議的標(biāo)識(shí)符，code () 抽象方法，主要表示的是業(yè)務(wù)指令，不同的業(yè)務(wù)對(duì)應(yīng)不同的指令。
@Data
public abstract class BaseBean implements Serializable {
private Integer tag=1;//固定值，標(biāo)識(shí)的是一個(gè)協(xié)議類型，不同協(xié)議對(duì)應(yīng)不同的值
public abstract Byte code();//業(yè)務(wù)指令抽象方法
}
登錄請(qǐng)求實(shí)體
@Data
public class LoginReqBean extends BaseBean implements Serializable {
private Integer userid;//用戶ID
private String username;//用戶名稱
public Byte code() {
return 1;//業(yè)務(wù)指令
}
}
登錄響應(yīng)實(shí)體
@Data
public class LoginResBean extends BaseBean implements Serializable {
private Integer status;//響應(yīng)狀態(tài)，0登錄成功，1登錄失敗
private String msg;//響應(yīng)信息
private Integer userid;//用戶ID
public Byte code() {
return 2;//業(yè)務(wù)指令
}
}
消息發(fā)送實(shí)體
public class MsgReqBean extends BaseBean implements Serializable {
private Integer fromuserid;//發(fā)送人ID
private Integer touserid;//接受人ID
private String msg;//發(fā)送消息
public Byte code() {
return 3;//業(yè)務(wù)指令
}
}
消息響應(yīng)響應(yīng)
public class MsgResBean extends BaseBean implements Serializable {
private Integer status;//響應(yīng)狀態(tài)，0發(fā)送成功，1發(fā)送失敗
private String msg;//響應(yīng)信息
public Byte code() {
return 4;//業(yè)務(wù)指令
}
}
消息接受實(shí)體
public class MsgRecBean extends BaseBean implements Serializable {
private Integer fromuserid;//發(fā)送人ID
private String msg;//消息
public Byte code() {
return 5;//業(yè)務(wù)指令
}
}
3.2 編碼和解碼實(shí)現(xiàn)
3.2.1 依賴坐標(biāo)
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.47</version>
</dependency>
3.2.2 編碼實(shí)現(xiàn)
public class MyEncoder extends MessageToByteEncoder<BaseBean> {
protected void encode(
ChannelHandlerContext channelHandlerContext,
BaseBean baseBean,
ByteBuf byteBuf) throws Exception {
//1.把實(shí)體序列化成字節(jié)數(shù)字
byte[] bytes= JSON.toJSONBytes(baseBean);
//2.根據(jù)協(xié)議組裝數(shù)據(jù)
byteBuf.writeInt(baseBean.getTag());//標(biāo)識(shí)（4個(gè)字節(jié)）
byteBuf.writeByte(baseBean.code());//指令（1個(gè)字節(jié)）
byteBuf.writeInt(bytes.length);//長(zhǎng)度（4個(gè)字節(jié)）
byteBuf.writeBytes(bytes);//
}
}
3.2.3 解碼實(shí)現(xiàn)
public class MyDecoder extends ByteToMessageDecoder {
protected void decode(
ChannelHandlerContext channelHandlerContext,
ByteBuf byteBuf,
List<Object> list) throws Exception {
//1.根據(jù)協(xié)議取出數(shù)據(jù)
int tag=byteBuf.readInt();//標(biāo)識(shí)符
byte code=byteBuf.readByte();//獲取指令
int len=byteBuf.readInt();//獲取數(shù)據(jù)長(zhǎng)度
byte[] bytes=new byte[len];
byteBuf.readBytes(bytes);
//2.根據(jù)code獲取類型
Class<? extends BaseBean> c= MapUtils.getBean(code);
//3.反序列化
BaseBean baseBean=JSON.parseObject(bytes,c);
list.add(baseBean);
}
}
3.2.4 指令和實(shí)體關(guān)系
為什么需要這么一個(gè)工具類呢？指令表示的是業(yè)務(wù)類型，不同的業(yè)務(wù)對(duì)應(yīng)不同的實(shí)體，那么解碼的時(shí)候，怎么知道反序列化成什么樣的實(shí)體呢？思路是獲取到的指令，再根據(jù)指令找到對(duì)應(yīng)的實(shí)體即可。
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.自定義一個(gè)Map，專門管理指令和實(shí)體的關(guān)系
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.根據(jù)指令獲取對(duì)應(yīng)的實(shí)體
public static Class<? extends BaseBean> getBean(Byte code){
try{
return map.get(code);
}catch (Exception e){
throw new RuntimeException(e.getMessage());
}
}
}
3.3 客戶端代碼實(shí)現(xiàn)
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.自定義業(yè)務(wù)
ch.pipeline().addLast(new ClientChatHandler());
//4.自定義編碼器
ch.pipeline().addLast(new MyEncoder());
}
});
3.3.2 業(yè)務(wù) Handler
所有的業(yè)務(wù)處理在同一個(gè) Handler 里面進(jìn)行處理，通過(guò)判斷實(shí)體類型來(lái)區(qū)分不同的業(yè)務(wù)處理。
public class ClientChatHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
//通道就緒時(shí)，發(fā)起登錄請(qǐng)求
login(ctx.channel());
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//根據(jù)msg做類型判斷，不同的業(yè)務(wù)做不同的處理
if(msg instanceof LoginResBean){
//1.登錄結(jié)果響應(yīng)
LoginResBean res=(LoginResBean) msg;
System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>登錄響應(yīng)："+res.getMsg());
if(res.getStatus()==0){
//1.登錄成功，則給通道綁定屬性
ctx.channel().attr(AttributeKey.valueOf("userid")).set(res.getUserid());
//2.調(diào)用發(fā)送消息方法
sendMsg(ctx.channel());
}else{
//1.登錄失敗，調(diào)用登錄方法
login(ctx.channel());
}
}else if(msg instanceof MsgResBean){
//1.發(fā)送消息結(jié)果響應(yīng)
MsgResBean res=(MsgResBean)msg;
System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>發(fā)送響應(yīng)："+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);
}
//發(fā)送消息方法
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(">>聊天內(nèi)容：");
String msg=scanner.next();
MsgReqBean bean=new MsgReqBean();
//從通道屬性獲取發(fā)送人ID
Integer fromuserid=(Integer) channel.attr(
AttributeKey.valueOf("userid")
).get();
//發(fā)送人ID
bean.setFromuserid(fromuserid);
//接受人ID
bean.setTouserid(touserid);
//發(fā)送消息
bean.setMsg(msg);
channel.writeAndFlush(bean);
}
}
}).start();
}
}
3.4 服務(wù)端代碼實(shí)現(xiàn)
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.業(yè)務(wù)Handler
ch.pipeline().addLast(new ServerChatHandler());
//4.自定義編碼器
ch.pipeline().addLast(new MyEncoder());
}
});
3.4.2 業(yè)務(wù) Handler
public class ServerChatHandler extends ChannelInboundHandlerAdapter{
//1.定義一個(gè)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.登錄請(qǐng)求
login((LoginReqBean) msg,ctx.channel());
}else if(msg instanceof MsgReqBean){
//2.發(fā)送消息請(qǐng)求
sendMsg((MsgReqBean)msg,ctx.channel());
}
}
//登錄處理方法
private void login(LoginReqBean bean, Channel channel){
LoginResBean res=new LoginResBean();
//從map里面根據(jù)用戶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.響應(yīng)
res.setStatus(0);
res.setMsg("登錄成功");
res.setUserid(bean.getUserid());
channel.writeAndFlush(res);
}else{
//通道不為空，證明該用戶已經(jīng)在線了
res.setStatus(1);
res.setMsg("該賬戶目前在線");
channel.writeAndFlush(res);
}
}
//消息發(fā)送處理方法
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. 小結(jié)
本節(jié)主要實(shí)現(xiàn)了單聊功能，實(shí)現(xiàn)思路相對(duì)還是稍微有點(diǎn)小復(fù)雜，大家主要核心掌握以下幾點(diǎn)思路：
- 不同業(yè)務(wù)設(shè)置不同的實(shí)體和指令，需要把指令和實(shí)體的關(guān)系管理起來(lái)，方便反序列化的時(shí)候，可以根據(jù)指令來(lái)反序列化到具體實(shí)體；
- 需要把用戶 ID 和通道的關(guān)系管理起來(lái)，并且靈活運(yùn)用 Channel 的 attr () 方法，該方法可以綁定屬性值，非常的有用。
2025 imooc.com All Rights Reserved |