1.引入RocketMQ依賴：首先，在pom.xml文件中添加RocketMQ的依賴：
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.0</version> <!-- 版本號(hào)根據(jù)實(shí)際情況調(diào)整 -->
</dependency>
2.配置RocketMQ連接信息：在application.properties或application.yml中配置RocketMQ的連接信息，包括Name Server地址等：
spring:
application:
name: ${sn.publish}
cloud:
stream:
rocketmq:
binder:
name-server: ${rocket-mq.name-server}
bindings:
output:
producer:
group: testSocket
sync: true
bindings:
output:
destination: test-topic
content-type: application/json3.消息發(fā)布組件
@Component
public class MqSourceComponent {
@Resource
Source source;
public void publishNotify(SampleNotifyDTO notify) {
source.output().send(MessageBuilder.withPayload(notify).build());
}
}4.消息發(fā)布控制器
@RestController
@Api(tags = "rocketmq")
public class MqController {
@Resource
MqSourceComponent mq;
@ApiOperation(value = "測(cè)試發(fā)布消息")
@PostMapping("test-publish")
public JsonVO<String> testSend(SampleNotifyDTO notify) {
mq.publishNotify(notify);
return JsonVO.success("消息已發(fā)送");
}
}項(xiàng)目結(jié)構(gòu)：
接下來(lái)是websocket模塊的搭建
1. 依賴添加
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.0</version> <!-- 版本號(hào)根據(jù)實(shí)際情況調(diào)整 -->
</dependency>
2.application.yml配置文件
server:
port: ${sp.ws}
spring:
application:
name: ${sn.ws}
cloud:
stream:
rocketmq:
binder:
name-server: ${rocket-mq.name-server}
bindings:
input:
destination: test-topic
content-type: application/json
group: testSocket3.將應(yīng)用程序綁定到消息代理
@EnableBinding(Sink.class): 這是Spring Cloud Stream的注解，它用于將應(yīng)用程序綁定到消息代理（如Kafka、RabbitMQ等）。Sink.class是Spring Cloud Stream提供的預(yù)定義輸入通道，允許你接收消息。通過(guò)這個(gè)注解，你的應(yīng)用程序可以監(jiān)聽消息通道，并定義消息處理邏輯。
@SpringBootApplication
@EnableDiscoveryClient
@EnableBinding(Sink.class)
public class WsApplication {
public static void main(String[] args) {
SpringApplication.run(WsApplication.class, args);
}
}4.消息訂閱組件
監(jiān)聽消息通道中的消息，一旦有消息到達(dá)，就會(huì)觸發(fā)listenNotify方法，該方法負(fù)責(zé)處理消息并通過(guò)chat服務(wù)發(fā)送響應(yīng)。
@Component
@Slf4j
public class MqListenComponent {
@Resource
ChatService chat;
@StreamListener(Sink.INPUT)
public void listenNotify(SampleNotifyDTO notify) {
log.info(notify.toString());
chat.sendMessage(notify.getClientId(), notify);
}
}5.消息通知服務(wù)
package com.zeroone.star.ws.service;
import cn.hutool.json.JSONUtil;
import lombok.SneakyThrows;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
@Component
@ServerEndpoint("/chat")
public class ChatService {
/**
* 連接會(huì)話池
*/
private static ConcurrentHashMap<String, Session> SESSION_POOL = new ConcurrentHashMap<>();
@OnOpen
public void onOpen(Session session) throws IOException {
// 判斷客戶端對(duì)象是否存在
if (SESSION_POOL.containsKey(session.getQueryString())) {
CloseReason closeReason = new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT, "ID沖突，連接拒絕");
session.getUserProperties().put("reason", closeReason);
session.close();
return;
}
// 將客戶端對(duì)象存儲(chǔ)到會(huì)話池
SESSION_POOL.put(session.getQueryString(), session);
System.out.println("客戶端（" + session.getQueryString() + "）：開啟了連接");
}
@OnMessage
public String onMessage(String msg, Session session) throws IOException {
// 解析消息 ==> ID::消息內(nèi)容
String[] msgArr = msg.split("::", 2);
// 處理群發(fā)消息，ID==all表示群發(fā)
if ("all".equalsIgnoreCase(msgArr[0])) {
for (Session one : SESSION_POOL.values()) {
// 排除自己
if (one == session) {
continue;
}
// 發(fā)送消息
one.getBasicRemote().sendText(msgArr[1]);
}
}
// 指定發(fā)送
else {
// 獲取接收方
Session target = SESSION_POOL.get(msgArr[0]);
if (target != null) {
target.getBasicRemote().sendText(msgArr[1]);
}
}
return session.getQueryString() + "：消息發(fā)送成功";
}
@OnClose
public void onClose(Session session) {
// 連接拒絕關(guān)閉會(huì)話
Object reason = session.getUserProperties().get("reason");
if (reason instanceof CloseReason) {
CloseReason creason = (CloseReason) reason;
if (creason.getCloseCode() == CloseReason.CloseCodes.CANNOT_ACCEPT) {
System.out.println("拒絕客戶（" + session.getQueryString() + "）：關(guān)閉連接");
return;
}
}
// 從會(huì)話池中移除會(huì)話
SESSION_POOL.remove(session.getQueryString());
System.out.println("客戶端（" + session.getQueryString() + "）：關(guān)閉連接");
}
@OnError
public void onError(Session session, Throwable throwable) {
System.out.println("客戶端（" + session.getQueryString() + "）錯(cuò)誤信息：" + throwable.getMessage());
}
@SneakyThrows
public void sendMessage(String id, Object message) {
// 群發(fā)
if ("all".equalsIgnoreCase(id)) {
for (Session one : SESSION_POOL.values()) {
// 發(fā)送消息
one.getBasicRemote().sendText(JSONUtil.toJsonStr(message));
}
}
// 指定發(fā)送
else {
// 獲取接收方
Session target = SESSION_POOL.get(id);
if (target != null) {
target.getBasicRemote().sendText(JSONUtil.toJsonStr(message));
}
}
}
}
項(xiàng)目結(jié)構(gòu)：文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-715115.html
文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-715115.html
到了這里，關(guān)于Sprint Cloud Stream整合RocketMq和websocket實(shí)現(xiàn)消息發(fā)布訂閱的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
