Sprint Cloud Stream整合RocketMq和websocket实现消息发布订阅

1.引入RocketMQ依赖:首先,在pom.xml文件中添加RocketMQ的依赖:

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.0</version> <!-- 版本号根据实际情况调整 -->
</dependency>

2.配置RocketMQ连接信息:在application.propertiesapplication.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/json

3.消息发布组件

@Component
public class MqSourceComponent {
    @Resource
    Source source;

    public void publishNotify(SampleNotifyDTO notify) {
        source.output().send(MessageBuilder.withPayload(notify).build());
    }
}

4.消息发布控制器

@RestController
@Api(tags = "rocketmq")
public class MqController {
    @Resource
    MqSourceComponent mq;

    @ApiOperation(value = "测试发布消息")
    @PostMapping("test-publish")
    public JsonVO<String> testSend(SampleNotifyDTO notify) {
        mq.publishNotify(notify);
        return JsonVO.success("消息已发送");
    }
}

项目结构:

接下来是websocket模块的搭建

1. 依赖添加

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.0</version> <!-- 版本号根据实际情况调整 -->
</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: testSocket

3.将应用程序绑定到消息代理

@EnableBinding(Sink.class): 这是Spring Cloud Stream的注解,它用于将应用程序绑定到消息代理(如Kafka、RabbitMQ等)。Sink.class是Spring Cloud Stream提供的预定义输入通道,允许你接收消息。通过这个注解,你的应用程序可以监听消息通道,并定义消息处理逻辑。

@SpringBootApplication
@EnableDiscoveryClient
@EnableBinding(Sink.class)
public class WsApplication {

    public static void main(String[] args) {
        SpringApplication.run(WsApplication.class, args);
    }

}

4.消息订阅组件

监听消息通道中的消息,一旦有消息到达,就会触发listenNotify方法,该方法负责处理消息并通过chat服务发送响应。

@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.消息通知服务

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 {
    /**
     * 连接会话池
     */
    private static ConcurrentHashMap<String, Session> SESSION_POOL = new ConcurrentHashMap<>();

    @OnOpen
    public void onOpen(Session session) throws IOException {
        // 判断客户端对象是否存在
        if (SESSION_POOL.containsKey(session.getQueryString())) {
            CloseReason closeReason = new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT, "ID冲突,连接拒绝");
            session.getUserProperties().put("reason", closeReason);
            session.close();
            return;
        }
        // 将客户端对象存储到会话池
        SESSION_POOL.put(session.getQueryString(), session);
        System.out.println("客户端(" + session.getQueryString() + "):开启了连接");
    }

    @OnMessage
    public String onMessage(String msg, Session session) throws IOException {
        // 解析消息 ==> ID::消息内容
        String[] msgArr = msg.split("::", 2);
        // 处理群发消息,ID==all表示群发
        if ("all".equalsIgnoreCase(msgArr[0])) {
            for (Session one : SESSION_POOL.values()) {
                // 排除自己
                if (one == session) {
                    continue;
                }
                // 发送消息
                one.getBasicRemote().sendText(msgArr[1]);
            }
        }
        // 指定发送
        else {
            // 获取接收方
            Session target = SESSION_POOL.get(msgArr[0]);
            if (target != null) {
                target.getBasicRemote().sendText(msgArr[1]);
            }
        }
        return session.getQueryString() + ":消息发送成功";
    }

    @OnClose
    public void onClose(Session session) {
        // 连接拒绝关闭会话
        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() + "):关闭连接");
                return;
            }
        }
        // 从会话池中移除会话
        SESSION_POOL.remove(session.getQueryString());
        System.out.println("客户端(" + session.getQueryString() + "):关闭连接");
    }

    @OnError
    public void onError(Session session, Throwable throwable) {
        System.out.println("客户端(" + session.getQueryString() + ")错误信息:" + throwable.getMessage());
    }

    @SneakyThrows
    public void sendMessage(String id, Object message) {
        // 群发
        if ("all".equalsIgnoreCase(id)) {
            for (Session one : SESSION_POOL.values()) {
                // 发送消息
                one.getBasicRemote().sendText(JSONUtil.toJsonStr(message));
            }
        }
        // 指定发送
        else {
            // 获取接收方
            Session target = SESSION_POOL.get(id);
            if (target != null) {
                target.getBasicRemote().sendText(JSONUtil.toJsonStr(message));
            }
        }
    }
}

项目结构:

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/107905.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

J2EE项目部署与发布(Windows版本)->会议OA单体项目Windows部署,spa前后端分离项目Windows部署

会议OA单体项目Windows部署spa前后端分离项目Windows部署 1.会议OA单体项目Windows部署&#xff08;以实施的角度&#xff09; 将项目放入webapp&#xff0c;项目能够访问: 首先拿到war包和数据库脚本&#xff0c;并检查是否有什么问题。 如何查看项目报错信息&#xff08;当你…

Nginx性能优化

简介 nginx作为常用的web代理服务器&#xff0c;某些场景下对于性能要求还是蛮高的&#xff0c;所以本片文章会基于操作系统调度以及网络通信两个角度来讨论一下Nginx性能的优化思路。 基于操作系统调度进行Nginx优化 CPU工作方式 对于用户进程&#xff0c;CPU会按照下面的…

深入浅出排序算法之堆排序

目录 1. 算法介绍 2. 执行流程⭐⭐⭐⭐⭐✔ 3. 代码实现 4. 性能分析 1. 算法介绍 堆是一种数据结构&#xff0c;可以把堆看成一棵完全二叉树&#xff0c;这棵完全二叉树满足&#xff1a;任何一个非叶结点的值都不大于(或不小于)其左右孩子结点的值。若父亲大孩子小&#x…

计算机操作系统重点概念整理-第三章 进程同步【期末复习|考研复习】

第三章 进程同步 【期末复习|考研复习】 计算机操作系统系列文章传送门&#xff1a; 第一章 计算机系统概述 第二章 进程管理 第三章 进程同步 第四章 内存管理 第五章 文件管理 第六章 输出输出I/O管理 文章目录 第三章 进程同步 【期末复习|考研复习】前言三、进程同步3.1 临…

C# 递归算法使用简介_常用整理

一、递归简介 递归算法是一种直接或者间接调用自身函数或者方法的算法。 递归算法的实质是把问题分解成规模缩小的同类问题的子问题&#xff0c;然后递归调用方法来表示问题的解。递归算法对解决一大类问题很有效&#xff0c;它可以使算法简洁和易于理解。 递归本质是循环&a…

Visual Studio Code的下载与安装

Visual Studio Code&#xff08;简称 VS Code&#xff09;是由 Microsoft 开发的免费、开源的文本编辑器&#xff0c;适用于多种操作系统&#xff0c;包括 Windows、macOS 和 Linux。它的设计目标是成为一款轻量级、高效的代码编辑工具&#xff0c;同时提供丰富的扩展和功能&am…

MySQL初始化之后启动报错(mysqld: Table ‘mysql.plugin‘ doesn‘t exist)

报错场景 初始化之后&#xff0c;服务无法启动。错误日志error-log 报错如下&#xff1a;&#xff08;mysql库下的系统表不存在&#xff09; 2023-10-26T06:03:08.150163-00:00 1 [System] [MY-013576] [InnoDB] InnoDB initialization has started. 2023-10-26T06:03:08.496…

Vite+Vue3项目全局引入scss文件

前言 Sass 是世界上最成熟、最稳定、最强大的专业级CSS扩展语言&#xff01;在日常项目开发过程中使用非常广泛&#xff0c;今天主要讲一下 ViteVue3 项目中该如何全局引入 scss 文件&#xff0c;引入混合 mixin 文件的不同配置。捎带说一下 Vue2 中的引入方式做一下简单的对比…

vue3从基础到入门(一)

文章目录 简介提升使用创建脚手架vite 常用Composition APIsetuprefreactive函数响应式vue2响应式vue3实现响应式 reactive对比ref注意计算属性computed函数 监视watch函数watchEffect函数 生命周期hook函数toRef 简介 2020年9月18日&#xff0c;Vue.js发布3.0版本&#xff0c…

PHP聊天系统源码 在线聊天系统网站源码 后台自适应PC与移动端

这个源码提供了前台和后台的自适应布局&#xff0c;可以在PC和移动端上完美展示。它支持一对多的交流&#xff0c;用户可以自由地创建新的房间并解散已创建的房间。 该程序还集成了签到功能和等级功能&#xff0c;让用户享受更多的互动乐趣。房间创建者具有禁言和拉黑用户的权…

LibreOffice编辑excel文档如何在单元格中输入手动换行符

用WPS编辑excel文档的时候&#xff0c;要在单元格中输入手动换行符&#xff0c;可以先按住Alt键&#xff0c;然后回车。 而用LibreOffice编辑excel文档&#xff0c;要在单元格中输入手动换行符&#xff0c;可以先按住Ctrl键&#xff0c;然后回车。例如&#xff1a;

K8s 部署 CNI 网络组件+k8s 多master集群部署+负载均衡

------------------------------ 部署 CNI 网络组件 ------------------------------ ---------- 部署 flannel ---------- K8S 中 Pod 网络通信&#xff1a; ●Pod 内容器与容器之间的通信 在同一个 Pod 内的容器&#xff08;Pod 内的容器是不会跨宿主机的&#xff09;共享同一…

node模块导出引入两种方式和npm包管理

模块化的好处 在 Node.js 中每个文件都被当做是一个独立的模块&#xff0c;模块内定义的变量和函数都是独立作用域的&#xff0c;因为 Node.js 在执行模块代码时&#xff0c;将使用如下所示的函数封装器对其进行封装 (function(exports,require,module,__filename,_dirname){//…

解密Kubernetes:探索开源容器编排工具的内核

&#x1f90d; 前端开发工程师&#xff08;主业&#xff09;、技术博主&#xff08;副业&#xff09;、已过CET6 &#x1f368; 阿珊和她的猫_CSDN个人主页 &#x1f560; 牛客高级专题作者、在牛客打造高质量专栏《前端面试必备》 &#x1f35a; 蓝桥云课签约作者、已在蓝桥云…

x210项目重新回顾之十七升级到linux4.19.114 +buildroot2018再讨论

代码参考https://github.com/colourfate/x210_bsp/ 他的是linux_4.10(dtb为 s5pv210-x210..dtb)我打算用linux4.19.114(dtb为 s5pv210-smdkv210.dtb) &#xff0c;所以修改build.sh ------------------------------------------------------------------------------ 5 M…

实用篇-认识微服务

一、服务架构演变 1. 单体架构 单体架构&#xff1a;将业务的所有功能集中在一个项目中开发&#xff0c;打成一个包部署 单体架构的优点&#xff1a; 架构简单部署成本低 单体架构的缺点&#xff1a; 耦合度高 2. 分布式架构 分布式架构&#xff1a; 根据业务功能对系…

京东平台数据分析(京东销量):2023年9月京东吸尘器行业品牌销售排行榜

鲸参谋监测的京东平台9月份吸尘器市场销售数据已出炉&#xff01; 根据鲸参谋电商数据分析平台的相关数据显示&#xff0c;今年9月&#xff0c;京东吸尘器的销量为19万&#xff0c;环比下滑约12%&#xff0c;同比下滑约25%&#xff1b;销售额为1.2亿&#xff0c;环比下滑约11%&…

MAC下安装Python

MAC基本信息&#xff1a; 执行命令&#xff1a; brew install cmake protobuf rust python3.10 git wget 遇到以下问题&#xff1a; > Downloading https://mirrors.aliyun.com/homebrew/homebrew-bottles/rust-1.59.0 Already downloaded: /Users/xxxx/Library/Caches/Ho…

vue笔记(三)

15、过滤器 过滤器 用法&#xff1a;对要显示的数据进行特定格式化后再显示&#xff08;用于一个简单的逻辑处理&#xff09;语法 1、注册过滤器&#xff1a;Vue.fillter(name,callback) &#xff08;全局&#xff09;或 new Vue{filters&#xff1a;{}}&#xff08;局部&…

FFmpeg5.1.3编译动态库踩坑之旅(基于Linux虚拟机)

准备工作 环境准备 1.Windows安装Oracle VM VirtualBox 7.0.10&#xff0c;安装ubuntu-22.04.3。 坑一&#xff1a;无法往虚拟机里拖放复制文件&#xff0c;解决办法&#xff1a;登录Ubuntu虚拟机时切换到xorg方式登录&#xff0c;参考地址&#xff1a;Ubuntu Desktop 22.04…