感覺(jué)Crossoverjie的一個(gè)開(kāi)源cim(即時(shí)通訊系統(tǒng)，源碼和設(shè)計(jì)個(gè)人覺(jué)得不錯(cuò)，空閑的時(shí)候分析一下。
cim github地址: https://github.com/crossoverJie/cim
協(xié)議設(shè)計(jì)
1. 請(qǐng)求協(xié)議類(lèi)圖
image.png
-
BaseRequest作為基類(lèi)，具有所有的請(qǐng)求都應(yīng)該具備的兩個(gè)屬性
public class BaseRequest {
// 請(qǐng)求的序列號(hào)
private String reqNo;
//請(qǐng)求的時(shí)間戳，構(gòu)造的時(shí)候就設(shè)置為System.currentTimeMillis() / 1000
private int timeStamp;
}
-
GoogleProtocolVO增加了requestId和msg兩個(gè)字段，表示傳輸GoogleProtocol 消息。 -
GroupReqVO增加了userId,msg兩個(gè)字段，表示傳輸?shù)氖侨毫南ⅰ?/li> -
LoginReqVO增加了userId,userName兩個(gè)字段，表示傳輸?shù)氖堑卿浵?/li> -
P2PReqVO增加了userId,receiveUserId,msg字段，表示傳輸?shù)氖且粚?duì)一私聊消息 -
SendMsgReqVO增加了msg,userId字段，表示通常的傳輸發(fā)送消息 -
StringReqVO增加了msg字段，表示用來(lái)傳輸String的消息
2. 相應(yīng)協(xié)議類(lèi)圖
image.png
-
CIMServerResVO用來(lái)接收查詢路由選中的服務(wù)器的響應(yīng)消息，格式如下：
{
code : 9000
message : 成功
reqNo : null
dataBody : {"ip":"127.0.0.1","port":8081}
}
-
OnlineUsersResVO用來(lái)接受查詢所有在線用戶的響應(yīng)消息，格式如下：
{
code : 9000
message : 成功
reqNo : null
dataBody : [{"userId":1545574841528,"userName":"zhangsan"},{"userId":1545574871143,"userName":"crossoverJie"}]
}
-
SendMsgResVO表示發(fā)送消息的響應(yīng)
3. 程序運(yùn)行流程
3.1 程序入口類(lèi)
public class CIMClientApplication implements CommandLineRunner{
private final static Logger LOGGER = LoggerFactory.getLogger(CIMClientApplication.class);
@Autowired
private ClientInfo clientInfo ;
public static void main(String[] args) {
SpringApplication.run(CIMClientApplication.class, args);
LOGGER.info("啟動(dòng) Client 服務(wù)成功");
}
@Override
public void run(String... args) throws Exception {
Scan scan = new Scan() ;
Thread thread = new Thread(scan);
thread.setName("scan-thread");
thread.start();
clientInfo.saveStartDate();
}
- 標(biāo)準(zhǔn)的Springboot啟動(dòng)流程，重寫(xiě)run方法在Springboot應(yīng)用啟動(dòng)后就啟動(dòng)一個(gè)線程去監(jiān)聽(tīng)控制臺(tái)，根據(jù)用戶的命令，做相應(yīng)的操作。
3.2 Scan掃描用戶的輸入命令
public void run() {
Scanner sc = new Scanner(System.in);
while (true) {
String msg = sc.nextLine();
//檢查消息,保證輸入消息不能不為null
if (msgHandle.checkMsg(msg)) {
continue;
}
//系統(tǒng)內(nèi)置命令
if (msgHandle.innerCommand(msg)){
continue;
}
//真正的發(fā)送消息
msgHandle.sendMsg(msg) ;
//寫(xiě)入聊天記錄
msgLogger.log(msg) ;
LOGGER.info("{}:【{}】", configuration.getUserName(), msg);
}
}
- 經(jīng)過(guò)檢查消息是否為空字符串，是否是內(nèi)置命令，最后剩下的是用戶發(fā)送的消息。
3.3 內(nèi)置命令的處理
如果是內(nèi)置命令，轉(zhuǎn)而通過(guò)反射實(shí)例化每個(gè)命令，這里用到命令模式。
public boolean innerCommand(String msg) {
if (msg.startsWith(":")) {
InnerCommand instance = innerCommandContext.getInstance(msg);
//調(diào)用里面的方法
instance.process(msg) ;
return true;
} else {
return false;
}
}
public InnerCommand getInstance(String command) {
//// 每個(gè)命令對(duì)應(yīng)一個(gè)實(shí)現(xiàn)類(lèi)
Map<String, String> allClazz = SystemCommandEnum.getAllClazz();
//兼容需要命令后接參數(shù)的數(shù)據(jù) :q cross
String[] trim = command.trim().split(" ");
String clazz = allClazz.get(trim[0]);
InnerCommand innerCommand = null;
try {
if (StringUtil.isEmpty(clazz)){
clazz = PrintAllCommand.class.getName() ;
}
//根據(jù)類(lèi)名獲取到在容器里面的實(shí)例
innerCommand = (InnerCommand) SpringBeanFactory.getBean(Class.forName(clazz));
} catch (Exception e) {
LOGGER.error("Exception", e);
}
return innerCommand;
}
-
內(nèi)部完整命令，以及他們的實(shí)現(xiàn)類(lèi)如下
image.png
完整命令類(lèi)類(lèi)圖如下：
image.png
看其中一個(gè)實(shí)現(xiàn)類(lèi)
public class PrintOnlineUsersCommand implements InnerCommand {
private final static Logger LOGGER = LoggerFactory.getLogger(PrintOnlineUsersCommand.class);
@Autowired
private RouteRequest routeRequest ;
@Override
public void process(String msg) {
try {
// 查詢所有的在線用戶，委托routeRequest 來(lái)查詢
List<OnlineUsersResVO.DataBodyBean> onlineUsers = routeRequest.onlineUsers();
LOGGER.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
for (OnlineUsersResVO.DataBodyBean onlineUser : onlineUsers) {
LOGGER.info("userId={}=====userName={}", onlineUser.getUserId(), onlineUser.getUserName());
}
LOGGER.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
} catch (Exception e) {
LOGGER.error("Exception", e);
}
}
}
都是通過(guò)其中的process來(lái)處理邏輯
3.4 處理內(nèi)置命令后，接著來(lái)看處理發(fā)送消息
public void sendMsg(String msg) {
if (aiModel) {
//ai 模式主要是調(diào)侃之前那個(gè)價(jià)值兩億的融資項(xiàng)目
aiChat(msg);
} else {
// 正常的聊天
normalChat(msg);
}
}
private void normalChat(String msg) {
String[] totalMsg = msg.split(";;");
// 私聊的格式是：12345;;hello
if (totalMsg.length > 1) {
//私聊
P2PReqVO p2PReqVO = new P2PReqVO();
p2PReqVO.setUserId(configuration.getUserId());
p2PReqVO.setReceiveUserId(Long.parseLong(totalMsg[0]));
p2PReqVO.setMsg(totalMsg[1]);
try {
p2pChat(p2PReqVO);
} catch (Exception e) {
LOGGER.error("Exception", e);
}
} else {
//群聊 直接發(fā)消息就行
GroupReqVO groupReqVO = new GroupReqVO(configuration.getUserId(), msg);
try {
groupChat(groupReqVO);
} catch (Exception e) {
LOGGER.error("Exception", e);
}
}
}
群聊和私聊也都委托 routeRequest來(lái)實(shí)現(xiàn)
@Override
public void groupChat(GroupReqVO groupReqVO) throws Exception {
routeRequest.sendGroupMsg(groupReqVO);
}
@Override
public void p2pChat(P2PReqVO p2PReqVO) throws Exception {
routeRequest.sendP2PMsg(p2PReqVO);
}
3.5 處理聊天記錄
接著最開(kāi)始的時(shí)候看，聊天完成后，需要把聊天記錄寫(xiě)入文件，實(shí)現(xiàn)如下
public void log(String msg) {
//開(kāi)始消費(fèi)，異步完成
startMsgLogger();
try {
//往阻塞隊(duì)列里面添加
blockingQueue.put(msg);
} catch (InterruptedException e) {
LOGGER.error("InterruptedException", e);
}
}
啟動(dòng)消息線程，往阻塞隊(duì)列里面添加消息
private class Worker extends Thread {
@Override
public void run() {
while (started) {
try {
//往阻塞隊(duì)列里面取
String msg = blockingQueue.take();
writeLog(msg);
} catch (InterruptedException e) {
break;
}
}
}
}
真正寫(xiě)入文件的實(shí)現(xiàn)如下：
private void writeLog(String msg) {
LocalDate today = LocalDate.now();
int year = today.getYear();
int month = today.getMonthValue();
int day = today.getDayOfMonth();
String dir = appConfiguration.getMsgLoggerPath() + appConfiguration.getUserName() + "/";
String fileName = dir + year + month + day + ".log";
Path file = Paths.get(fileName);
boolean exists = Files.exists(Paths.get(dir), LinkOption.NOFOLLOW_LINKS);
try {
if (!exists) {
Files.createDirectories(Paths.get(dir));
}
List<String> lines = Arrays.asList(msg);
Files.write(file, lines, Charset.forName("UTF-8"), StandardOpenOption.CREATE, StandardOpenOption.APPEND);
} catch (IOException e) {
LOGGER.info("IOException", e);
}
}
查找聊天記錄的實(shí)現(xiàn)如下，就是簡(jiǎn)單的查找每個(gè)文件的每行，然后看是否包含，這樣的方式很暴力，后期的話有很大改進(jìn):
@Override
public String query(String key) {
StringBuilder sb = new StringBuilder();
Path path = Paths.get(appConfiguration.getMsgLoggerPath() + appConfiguration.getUserName() + "/");
try {
Stream<Path> list = Files.list(path);
List<Path> collect = list.collect(Collectors.toList());
for (Path file : collect) {
List<String> strings = Files.readAllLines(file);
for (String msg : strings) {
if (msg.trim().contains(key)) {
sb.append(msg).append("\n");
}
}
}
} catch (IOException e) {
LOGGER.info("IOException", e);
}
return sb.toString().replace(key, "\033[31;4m" + key + "\033[0m");
}
3.6 RouteRequestImpl的實(shí)現(xiàn)
這個(gè)實(shí)現(xiàn)里面包含眾多的功能，例如，群聊，私聊，離線，獲取在線用戶，獲取一個(gè)可用的服務(wù)ip。這些功能的實(shí)現(xiàn)都是依靠 RouteRequestImpl來(lái)完成，而RouteRequestImpl里面的實(shí)現(xiàn)是通過(guò)okhttp遠(yuǎn)程調(diào)用cim-router的http接口實(shí)現(xiàn)的?？雌渲械娜毫墓δ埽?/p>
public void sendGroupMsg(GroupReqVO groupReqVO) throws Exception {
//序列化
JSONObject jsonObject = new JSONObject();
jsonObject.put("msg",groupReqVO.getMsg());
jsonObject.put("userId",groupReqVO.getUserId());
RequestBody requestBody = RequestBody.create(mediaType,jsonObject.toString());
Request request = new Request.Builder()
.url(groupRouteRequestUrl)
.post(requestBody)
.build();
//發(fā)送http請(qǐng)求cim-router
Response response = okHttpClient.newCall(request).execute() ;
try {
if (!response.isSuccessful()){
throw new IOException("Unexpected code " + response);
}
}finally {
response.body().close();
}
}
3.7 客戶端的啟動(dòng)
上面所有的都是內(nèi)置命令的處理以及和cim-router的通信。但是，client最終是要和server通信的，所以在這個(gè)過(guò)程中，客戶端作為netty客戶端需要啟動(dòng)。這個(gè)啟動(dòng)過(guò)程可以在CIMClient實(shí)例化的過(guò)程中啟動(dòng)
@Component
public class CIMClient {
//構(gòu)造函數(shù)完成后調(diào)用
@PostConstruct
public void start() throws Exception {
//登錄 + 獲取可以使用的服務(wù)器 ip+port
CIMServerResVO.ServerInfo cimServer = userLogin();
//啟動(dòng)客戶端
startClient(cimServer);
//向服務(wù)端注冊(cè)
loginCIMServer();
}
}
向路由注冊(cè)并返回可用的服務(wù)器地址
private CIMServerResVO.ServerInfo userLogin() {
LoginReqVO loginReqVO = new LoginReqVO(userId, userName);
CIMServerResVO.ServerInfo cimServer = null;
try {
//獲取可用的服務(wù)器
cimServer = routeRequest.getCIMServer(loginReqVO);
//保存系統(tǒng)信息
clientInfo.saveServiceInfo(cimServer.getIp() + ":" + cimServer.getCimServerPort())
.saveUserInfo(userId, userName);
LOGGER.info("cimServer=[{}]", cimServer.toString());
} catch (Exception e) {
errorCount++;
if (errorCount >= configuration.getErrorCount()) {
LOGGER.error("重連次數(shù)達(dá)到上限[{}]次", errorCount);
msgHandle.shutdown();
}
LOGGER.error("登錄失敗", e);
}
return cimServer;
}
啟動(dòng)客戶端到服務(wù)端（上一步獲取的）的channel
private void startClient(CIMServerResVO.ServerInfo cimServer) {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new CIMClientHandleInitializer())
;
ChannelFuture future = null;
try {
future = bootstrap.connect(cimServer.getIp(), cimServer.getCimServerPort()).sync();
} catch (InterruptedException e) {
errorCount++;
if (errorCount >= configuration.getErrorCount()) {
LOGGER.error("鏈接失敗次數(shù)達(dá)到上限[{}]次", errorCount);
msgHandle.shutdown();
}
LOGGER.error("連接失敗", e);
}
if (future.isSuccess()) {
LOGGER.info("啟動(dòng) cim client 成功");
}
channel = (SocketChannel) future.channel();
}
向服務(wù)器注冊(cè)
private void loginCIMServer() {
CIMRequestProto.CIMReqProtocol login = CIMRequestProto.CIMReqProtocol.newBuilder()
.setRequestId(userId)
.setReqMsg(userName)
.setType(Constants.CommandType.LOGIN)
.build();
ChannelFuture future = channel.writeAndFlush(login);
future.addListener((ChannelFutureListener) channelFuture ->
LOGGER.info("注冊(cè)成功={}", login.toString()));
}
總結(jié)
到這里整cim-client的功能就完成了，客戶端就是通過(guò)命令模式通過(guò)okhttp遠(yuǎn)程調(diào)用特定的服務(wù)地址來(lái)注冊(cè)，獲取服務(wù)器地址，完成運(yùn)維。通過(guò)從cim-router拿到的服務(wù)器地址，建立客戶端-服務(wù)端的連接，即可完成消息私聊，群聊。
