前言
其實關於這個的話，我先前的幾篇博文： SpringBoot+Netty+Vue+Websocket實現線上推送/聊天系統
實用水文篇——SpringBoot整合Netty實現訊息推送伺服器
其實已經說的非常明白了，但是每想到後臺還是有人找我要程式碼，因為完整的程式碼其實在博文裡面都是貼出來了的，當然前面的博文我都是給一個類似於腳手架的東西，沒有給出實際的案例。那麼今天也是直接給出程式碼的案例吧，這裡再次宣告，所有的程式碼都是在這兩篇博文裡面有的，如果做了修改在本文當中會給出提示。此外的話，我們的專案是完全開源的，但是在開發階段不開源，因為有一些銘感資訊。在開發完成之後開源，這裡請見諒，當然也是為什麼我沒有辦法給出原始碼的原因（暫時）也是為什麼你可以在我的那兩篇博文看到完整的程式碼資訊的原因。
Tips： 在我的頻道只會告訴你怎麼做飯，不會把飯菜做好了再給你，如果需要，那是另外“價格”。
那麼廢話不多說，我們開始。
技術架構
我們今天來看到我們的一個案例。首先是我們的技術架構：
那麼在我們今天的話是這樣的：
使用者上傳博文
博文透過之後傳送稽核訊息給前端
前端對訊息進行展示
效果圖
這個是我們上傳博文，博文透過之後會看到訊息有個提示。
之後到具體的頁面可以看到訊息
因為圖片是中午擷取的，有個小bug沒有調整，但是不要在意這些，這個bug是很簡單的，因為一開始我這邊做過初始化，沒有清除快取罷了。
後端專案
之後的話來看到我們的後端的一個服務情況。
我們要進行互動的服務就是這三個，一個是閘道器，一個是博文的服務，還有就是我們的訊息伺服器。因為我們是使用者上傳成功後再顯示的。
那麼關於部落格的模組的話在這塊有完整的說明： SpringBoot + Vue實現博文上傳+展示+博文列表
我們這邊做的就是一個系列。當然只是演示實際上，你用先前我給出的程式碼是完全可以實現效果的，我們這邊只是說用那一套程式碼來真正做我們具體的業務。
訊息資料定義
儲存結構
那麼在開始之前的話，我們這邊對我們的訊息伺服器設計了對應的資料庫用來儲存訊息。
這一塊看你自己，我們這邊肯定是要的。
那麼我們這次的案例需要使用到的表是這個表：
訊息狀態
之後的話，我們需要對我們的訊息進行定義。 我們在這裡對訊息的狀態做出如下定義：
訊息具備兩者狀態，針對兩個情況
簽收狀態，即，伺服器確定使用者線上，並且將訊息傳輸到了客戶端，為簽收狀態。
閱讀狀態，在保證已簽收的情況下，使用者是否已經閱讀訊息，這部分的邏輯有客戶端程式碼處理。
對應未簽收的訊息，使用者上線時，請求伺服器是否存在未簽收的訊息，如果有，進行統一讀取，儲存到本地
對於未讀訊息，主要是對使用者的狀態進行一個判斷，訊息已經快取到使用者本地。
那麼此時的話，我們就已經說清楚了這個。在我們的資料庫裡面status這個欄位就是用來判斷使用者是不是簽收了訊息的。至於使用者到底有沒有讀取訊息，那麼完全就是客戶端需要做的判斷了。
當然你也可以設計為全部由服務端來處理。
Nutty訊息服務
專案結構
ok，說完了這個的話，我們再來看到我們的訊息服務端是怎麼處理的。
首先我們依然是和先前的博文一樣，保留了先前的東西。 但是我們這邊多了Controller，和Dao層。
那麼在這邊的話，我們需要關注的只有這幾個東西：
這幾個東西就是我們來實現前面的效果的實際的業務程式碼。
除了這些當然還有我們的Dao，但是這個是根據你的業務來的，這裡我就不展示了，類比嘛。
改動
那麼說完了這些，我們來看到和先前的程式碼有哪些改動的東西。
訊息bean
首先是我們的訊息的改動。
@AllArgsConstructor@NoArgsConstructor@ToString/** * 由於我們這邊Nutty處理的訊息只有註冊，所以話這裡只需要 * 保留action和userid即可 * */public class DataContent implements Serializable { private Integer action； private String userid；}複製程式碼
那麼我們的訊息的型別是這樣的：
public enum MessageActionEnum { //定義訊息型別 CONNECT（1，“第一次（或重連）初始化連線”）， CHAT（2，“聊天訊息”）， SIGNED（3，“訊息簽收”）， KEEPALIVE（4，“客戶端保持心跳”）， PULL_FRIEND（5， “拉取好友”）， HOLEADUITMSG（6，“稽核訊息”）； public final Integer type； public final String content； MessageActionEnum（Integer type，String content） { this。type = type； this。content = content； }}複製程式碼
訊息處理器
既然我們的這個訊息型別變了，那麼我們的這個程式碼也變了：
@Component@ChannelHandler。Sharablepublic class ServerListenerHandler extends SimpleChannelInboundHandler
這個就是我們核心的訊息處理器。
那麼其他的關於Nutty的玩意我壓根沒有改動。
訊息轉換pojo工具
這裡還有咱們的訊息轉換的工具類。這個的話，我也給一下：
public class JsonUtils { // 定義jackson物件 private static final ObjectMapper MAPPER = new ObjectMapper（）； /** * 將物件轉換成json字串。 *
Title： pojoToJson
*Description：
* @param data * @return */ public static String objectToJson（Object data） { try { String string = MAPPER。writeValueAsString（data）； return string； } catch （JsonProcessingException e） { e。printStackTrace（）； } return null； } /** * 將json結果集轉化為物件 * * @param jsonData json資料 * @param beanType 物件型別 * @return */ public staticTitle： jsonToList
*Description：
* @param jsonData * @param beanType * @return */ public static前面應該是有給的。
重點就是咱們的這個Controller這一塊。
稽核訊息處理
controller
我們首先來看到我們的Controller
@RestController@RequestMapping（“/message/holeAduit”）public class HoleAduitMsgController { @Autowired HoleAduitMsgService holeAduitMsgService； @PostMapping（“/aduit”） public R holeAduitMsg（@Validated @RequestBody HoleAduitMsgQ holeAduitMsgQ）{ return holeAduitMsgService。holeaduitMsg（holeAduitMsgQ）； }}複製程式碼
我們只看到這一個介面，因為其他的都是類似的。
那麼這裡的話我們還是需要一個請求的實體類的。 那麼這個實體類的話是這個樣子的：
@Data@AllArgsConstructor@NoArgsConstructorpublic class HoleAduitMsgQ { @NotEmpty private String userid； private String msg； private String msgtitle； private Long linkid； private Integer type；}複製程式碼
這個實體類的話，是被我封裝了這裡：
因為我們是一個微服務，所以的話，對應的這個請求我們都是放在了第三方的一個包下面。 那麼對於的還有咱們暴露出來的服務。
@FeignClient（“message”）@RequestMapping（“/message/holeAduit”）public interface FeignHoleAduitMsgService { @PostMapping（“/aduit”） public R holeAduitMsg（@RequestBody HoleAduitMsgQ holeAduitMsgQ）；}複製程式碼
之後的話，我們可以看到具體的實現類。
實現類
@Servicepublic class HoleAduitMsgServiceImpl implements HoleAduitMsgService { @Autowired HoleAuditService auditService； @Override public R holeaduitMsg（HoleAduitMsgQ holeAduitMsgQ） { //1。對訊息進行儲存，只要使用者線上的話，我們就直接先給他簽收一下 String userid = holeAduitMsgQ。getUserid（）； Channel channel = UserConnectPool。getChannelFromMap（userid）； HoleAuditEntity holeAuditEntity = new HoleAuditEntity（）； BeanUtils。copyProperties（holeAduitMsgQ，holeAuditEntity）； holeAuditEntity。setCreateTime（DateUtils。getCurrentTime（））； if（channel！=null）{ //這邊只是保證存在，雙層保險，這個時候的話就是線上 Channel realChannel = UserConnectPool。getChannelGroup（）。find（channel。id（））； if（realChannel！=null）{ holeAuditEntity。setStatus（1）； //我們這邊直接轉發訊息就好了，不需要再額外處理 realChannel。writeAndFlush（ new TextWebSocketFrame（ JsonUtils。objectToJson（ Objects。requireNonNull（R。ok（）。put（“data”， holeAuditEntity）） 。put（“type”， MessageActionEnum。HOLEADUITMSG。type） ） ） ）； } } //這裡進行訊息的儲存 auditService。save（holeAuditEntity）； return R。ok（）； }}複製程式碼
這裡面的邏輯其實非常簡單，就幾個步驟。
1。接受請求 2。判斷使用者是否線上，線上推送，並儲存設定為已簽收（訊息） 如果不線上，不進行推送，但是儲存訊息並設定為未簽收
這裡的話就是非常簡單的。
服務呼叫
之後的話，就是我們的呼叫。我們的呼叫是在我們的部落格服務進行呼叫的。
我們先看到我們完整的部落格服務的實現類。
public class BlogUpServiceImpl implements BlogUpService { @Autowired FeignUserService feignUserService； @Autowired ContentService contentService； @Autowired FeignHeadimgService feignHeadimgService； @Autowired WordFilter wordFilter； @Autowired BlogService blogService； @Autowired FeignLogActicleService feignLogActicleService； @Autowired RedisUtils redisUtils； @Autowired FeignHoleAduitMsgService feignHoleAduitMsgService； private final static Double threshold = 0。05； /** * 介面對使用者進行十分鐘限制 * 1。完成使用者博文的上傳 * 2。儲存使用者博文，博文對應資訊 * 3。修改使用者日誌 * */ @Override public R blogUp（UpBlogEntity entity） { String userid = entity。getUserid（）； String backMessage = “success”； //介面限流 if（redisUtils。hasKey（RedisTransKey。getBlogUpKey（entity。getUserid（））））{ return R。error（BizCodeEnum。OVER_UPBLOG。getCode（）， BizCodeEnum。OVER_UPBLOG。getMsg（））； } R info = feignUserService。info（userid）； String userString = FastJsonUtils。toJson（info。get（“user”））； UserEntity user = FastJsonUtils。fromJson（userString， UserEntity。class）； if（user！=null）{ String context = entity。getContext（）； String blogInfo = entity。getInfo（）； /** * 先對context和bloginfo進行校驗，是否為存在不友好的資訊 * */ int countContext = wordFilter。wordCount（context）； int countInfo = wordFilter。wordCount（blogInfo）； int status = 1； //博文的摘要過濾，只要摘要沒有過，直接先打回去！ if（countInfo>=blogInfo。length（）*threshold）{ return R。error（BizCodeEnum。BAD_BLOGINFO。getCode（），BizCodeEnum。BAD_BLOGINFO。getMsg（））； } //博文內容的過濾 if（countContext>=context。length（）*threshold）{ //直接就是沒有透過稽核 return R。error（BizCodeEnum。BAD_CONTEXT。getCode（），BizCodeEnum。BAD_CONTEXT。getMsg（））； }else if （countContext>0&&countContext 這裡面有註釋，重點就是那個傳送訊息的。 到此的話，我們的後端就沒啥事情了。 前端 之後的話就是我們的前端。 我們的前端主要是負責兩件事情 向伺服器註冊（如果使用者登入了的話） 保持心跳線上 接收伺服器傳送過來的訊息，並儲存，然後要通知使用者 連線程式碼 首先是我們先前博文有說到的，我們的連線封裝好的程式碼。 我這裡是放在了socket包下面，其他的你們自己看著辦。 // 匯出socket物件export { socket}import { Message } from ‘element-ui’// socket主要物件var socket = { websock： null， /** * 這個是我們的ws的地址 * */ ws_url： “ws：//localhost：9000/ws”， userid： null， msgfunc： null， /** * 開啟標識 * */ socket_open： false， /** * 心跳timer * */ hearbeat_timer： null， /** * 心跳傳送頻率 * */ hearbeat_interval： 10000， /** * 是否開啟重連 * */ is_reonnect： true， /** * 重新連線的次數 * */ reconnect_count： 3， /** * 當前重新連線的次數，預設為：1 * */ reconnect_current： 1， /** * 重新連線的時間型別 * */ reconnect_timer： null， /** * 重新連線的間隔 * */ reconnect_interval： 3000， /** * 登入後才進行連線 * */ /** * 初始化連線 */ init： （） => { let loginToken = localStorage。getExpire（“LoginToken”）； let userid = localStorage。getExpire（“userid”）； if（loginToken==null && userid==null） { Message（{ message： ‘當前正在以遊客身份訪問’， type： ‘info’， }）； return ； } if （！（“WebSocket” in window）） { Message（{ message： ‘當前瀏覽器與網站不相容丫’， type： ‘error’， }）； console。log（‘瀏覽器不支援WebSocket’） return null } // 已經建立過連線不再重複建立 if （socket。websock） { return socket。websock } socket。websock = new WebSocket（socket。ws_url） socket。websock。onmessage = function （e） { socket。receive（e） } // 關閉連線 socket。websock。onclose = function （e） { console。log（‘連線已斷開’） console。log（‘connection closed （’ + e。code + ‘）’） clearInterval（socket。hearbeat_interval） socket。socket_open = false // 需要重新連線 if （socket。is_reonnect） { socket。reconnect_timer = setTimeout（（） => { // 超過重連次數 if （socket。reconnect_current > socket。reconnect_count） { clearTimeout（socket。reconnect_timer） return } // 記錄重連次數 socket。reconnect_current++ socket。reconnect（） }， socket。reconnect_interval） } } // 連線成功 socket。websock。onopen = function （） { Message（{ message： ‘Welcome here’， type： ‘success’， }）； let userid = localStorage。getExpire（“userid”）； socket。userid = userid； console。log（‘連線成功’） socket。socket_open = true socket。is_reonnect = true // 開啟心跳 socket。heartbeat（） //註冊使用者 let resit={ “action”： 1， “userid”： userid } socket。send（resit） } // 連線發生錯誤 socket。websock。onerror = function （err） { Message（{ message： ‘無法連線至伺服器！’， type： ‘error’， }）； console。log（‘WebSocket連線發生錯誤’） } }， /** * 獲取websocket物件 * */ getSocket：（）=>{ //建立了直接返回，反之重來 if （socket。websock） { return socket。websock }else { socket。init（）； } }， getStatus：（）=> { if （socket。websock。readyState === 0） { return “未連線”； } else if （socket。websock。readyState === 1） { return “已連線”； } else if （socket。websock。readyState === 2） { return “連線正在關閉”； } else if （socket。websock。readyState === 3） { return “連線已關閉”； } }， /** * 傳送訊息 * @param {*} data 傳送資料 * @param {*} callback 傳送後的自定義回撥函式 */ send： （data， callback = null） => { // 開啟狀態直接傳送 if （socket。websock。readyState === socket。websock。OPEN） { socket。websock。send（JSON。stringify（data）） if （callback） { callback（） } // 正在開啟狀態，則等待1s後重新呼叫 } else if （socket。websock。readyState === socket。websock。CONNECTING） { setTimeout（function （） { socket。send（data， callback） }， 1000） // 未開啟，則等待1s後重新呼叫 } else { socket。init（） setTimeout（function （） { socket。send（data， callback） }， 1000） } }， /** * 接收訊息 * @param {*} message 接收到的訊息 */ receive： （message） => { var recData = JSON。parse（message。data） /** *這部分是我們具體的對訊息的處理 * */ if（socket。msgfunc==null）{ Message（{ message： ‘receive需要傳入一個func進行全域性訊息處理！’， type： ‘error’， }）； }else { socket。msgfunc（recData） } }， /** * 心跳 */ heartbeat： （） => { console。log（‘socket’， ‘ping’） if （socket。hearbeat_timer） { clearInterval（socket。hearbeat_timer） } socket。hearbeat_timer = setInterval（（） => { //傳送心跳包 let data = { “action”： 4， “userid”： socket。userid } socket。send（data） }， socket。hearbeat_interval） }， /** * 主動關閉連線 */ close： （） => { console。log（‘主動斷開連線’） clearInterval（socket。hearbeat_interval） socket。is_reonnect = false socket。websock。close（） }， /** * 重新連線 */ reconnect： （） => { console。log（‘發起重新連線’， socket。reconnect_current） if （socket。websock && socket。socket_open） { socket。websock。close（） } socket。init（） }，}複製程式碼 這段程式碼裡面主要有兩個重要的地方、 初始化的連線 傳送訊息（這邊的話我們基本上是透過http傳送到具體的服務，然後由訊息伺服器轉發的，但是心跳線上還是需要這個的） 接收到訊息的處理方法（這個要自己實現的，怎麼實現待會會有說明） 初始化 首先是初始化，我這邊的話是做訊息的推送，包括使用者的聊天，所以的話，我們這邊是需要全域性使用的。那麼這邊的話，我們初始化需要在你的根頁面進行使用。那麼我這邊是在這邊在這裡。你們自己的自己看著辦。 接受訊息 之後是我們接受訊息的地方。 哪裡需要哪裡使用。我這邊是在這個地方使用： 這部分的完整程式碼是這樣的： created（） { /** * 在這裡我們負責對訊息進行處理，我們把訊息儲存到快取當中 * 到具體的頁面的時候，我們就載入，這樣就好了。 * */ socket。msgfunc=this。msgfunc； //載入訊息 let messageNum_Local = localStorage。getExpire（“messageNum”）； let messageContent_Local = localStorage。getExpire（“messageContent”）； if（messageNum_Local）{ this。messageNum = messageNum_Local； this。total = messageNum_Local。total； }else { this。messageNum = messageNum； localStorage。setExpire（“messageNum”， this。messageNum，this。OverTime）； this。total = 0； } if（messageContent_Local）{ this。messageContent = messageContent_Local； }else { this。messageContent = messageContent； //因為一開始有初始值，這個初始值是不要的 delete this。messageContent［0］； localStorage。setExpire（“messageContent”，this。messageContent，this。OverTime）； } }，複製程式碼 msgfunc（res）{ //這個msgfunc是負責處理全域性資訊的 if（res。type===4）{ //心跳正常 }else { /** * 這裡面就是我們接下來要做的邏輯了， * 這裡的話訊息是分為兩個狀態的，簽收和讀取 * */ let messageNum_Local = localStorage。getExpire（“messageNum”）； messageNum_Local。total+=1； this。total = messageNum_Local。total； this。messageNum = messageNum_Local； //載入訊息 this。messageContent = localStorage。getExpire（“messageContent”）； if（res。type===6）{ //這個時候是我們的稽核訊息 this。messageNum。auditNum+=1； //頭部插入訊息 this。messageContent。auditContent。unshift（res。data） } //當token過期的時候，我們這些東西都需要進行刪除 localStorage。setExpire（“messageNum”，this。messageNum，this。OverTime）； localStorage。setExpire（“messageContent”，this。messageContent，this。OverTime）； } }，複製程式碼 我們在這個頁面實現了對訊息的處理，並且讓我們的socket進行了指定。 那麼這段程式碼主要做了這些事情 指定了訊息的處理方法（初始化部分） 在訊息處理部分，主要是針對訊息的型別，進行不同的儲存 我們著這邊定義了兩個訊息儲存的玩意。 這個就是我們定義儲存訊息的玩意 let messageNum = { total： 0， auditNum： 0，}；let messageContent = { auditContent：［ { userid： null， msg： null， msgtitle： null， linkid： null， createTime： null， msgid： null， status： null， type： null， } ］}export { messageNum， messageContent}複製程式碼 剛剛的那一段程式碼就是為了儲存訊息。這裡的話，目前是隻定義了一個訊息型別。 訊息的展示 盡然我們把訊息儲存起來了，那麼我們就需要進行讀取了。 那麼讀取的話很簡單啊，我們都存起來了，直接讀取載入不就好了。 那麼在這邊的話，就是咱們的這個頁面 檢視詳情：{{message。msg}} {{message。createTime}} 這塊的實現現在還是很粗糙，反正就是一個案例。這下子的話完整的過程我是已經說明白了的。 其實如果有時間的話，完全是可以把這個封裝成一個專門的訊息處理中介軟體的。剛好有用，而且很有必要。這裡的話，我可以有機會試試，仿造一下，也搞一個訊息中介軟體玩玩。 效果 之後的 話就是我們的測試效果。 登入有提示： 這邊有心跳包： 後端也可以看到連線 編寫文章的時候，上傳成功後有訊息提示： 這個是我們上傳博文，博文透過之後會看到訊息有個提示。 然後還可以到具體的頁面檢視。 總結 這裡的話，如果這個系列ok了的話，那麼此時你已經學會了 CURD+Nutty了。此時你已經可以脫離XXX管理系統了，可以做一個微信聊天之類的東西了。那麼在我們這邊的不足之處的話，還差一個就是聊天的資訊加密，但是這個東西，怎麼說呢，如果是類似於md5這種型別的加密，那麼是不可能有人破解，但是我也不可能還原，我還原不了，你也就只能看到亂碼。我要是可以解密，那麼只是保證了在傳輸的時候是安全的，但是我一樣可以看，除非聊天秘鑰在你手裡，但是你的秘鑰總還是要再我們的伺服器儲存的，至少是要上傳的，不然我怎麼給你解密。所以這個我可能都不會去加密，頂多傳輸的時候加一下，甚至不加上一個https就完了。 作者：the_way_inf 連結：https：//juejin。cn/post/7159512307398574087
