java springboot+rabbitmq+websocket 订阅展示

记录工作

需要的依赖

<!--fastjson坐标-->

        <dependency>

            <groupId>com.alibaba</groupId>

            <artifactId>fastjson</artifactId>

            <version>1.2.79</version>

        </dependency>

        <!--RabbitMQ-->

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-amqp</artifactId>

        </dependency>

        <!--WebSocket-->

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-websocket</artifactId>

        </dependency>

        <dependency>

            <groupId>com.alibaba</groupId>

            <artifactId>fastjson</artifactId>

            <version>1.2.46</version>

        </dependency>

        <!--JSON-->

        <dependency>

            <groupId>net.sf.json-lib</groupId>

            <artifactId>json-lib</artifactId>

            <version>2.4</version>

            <classifier>jdk15</classifier>

        </dependency>

rabbitmq:

    host: 127.0.0.0.1  #（自己的服务地址）

    port: 5672

    username: admin  #rabbitmq的用户名

    password: admin  #密码

    virtual-host: /

      # 开启消息发送确认

    publisher-confirm-type: correlated

    publisher-returns: true

    listener:

      simple:

        acknowledge-mode: manual

rabbitmq创建队列以及路由配置类，这里用的是路由的方式

package com.example.demotest.config;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.*;

import org.springframework.amqp.rabbit.connection.ConnectionFactory;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

/**

 * @title

 * @description RabbitMQ 配置类

 * @author admin  issuser

 * @updateTime 2022/11/10 15:44

 * @throws

 */

@Slf4j

@Configuration

public class RabbitConfig {

    //websocket 消息队列

    public static final String msg_queue = "msg_queue";

    //消息交换机

    public static final String msg_exchang = "msg_exchang";

    //消息路由键

    public static final String msg_routing_key = "msg_routing_key";

    /**

     * 消息队列

     * @return

     */

    @Bean

    public Queue msgQueue(){

        return new Queue(msg_queue);

    }

    @Bean

    public Queue newQueue(){

        return new Queue("new_queue");

    }

    @Bean

    public DirectExchange directExchange(){

        return new DirectExchange(msg_exchang);

    }

    @Bean

    public DirectExchange directExchange1(){

        return new DirectExchange("new_exchang");

    }

    /**

     * 消息队列绑定消息交换机

     * @return

     */

    @Bean

    public Binding msgBinding(){

        return BindingBuilder.bind(msgQueue()).to(directExchange()).with(msg_routing_key);

    }

    @Bean

    public Binding meBinding(){

        return BindingBuilder.bind(newQueue()).to(directExchange1()).with("me_key");

    }

    @Bean

    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){

        RabbitTemplate rabbitTemplate = new RabbitTemplate();

        rabbitTemplate.setConnectionFactory(connectionFactory);

        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数

        rabbitTemplate.setMandatory(true);

        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> log.info("发送订阅消息确认，确认情况：{}, 原因：{}", ack, cause));

        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> log.info("发送到订阅消息: {} , 回应码: {}, 回应信息: {}, 交换机: {}, 路由键: {}", message, replyCode, replyText, exchange, routingKey));

        return rabbitTemplate;

    }

}

发送消息的配置类，我这里稍微改造了一下直接变成了接口，如果不需要只需要删除接口有关的注解就可以了，返回类型其实是void

package com.example.demotest.controller;

import com.alibaba.fastjson.JSONObject;

import com.example.demotest.config.RabbitConfig;

import com.example.demotest.entity.ConfigurationTable;

import com.example.demotest.util.Redis;

import io.swagger.annotations.Api;

import io.swagger.annotations.ApiOperation;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.rabbit.connection.CorrelationData;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

import java.util.UUID;

/**

 * @title

 * @description 消息提供者

 * @author admin  issuser

 * @updateTime 2022/11/10 15:44

 * @throws

 */

@Slf4j

@Component

@RestController

@Api(tags = "发消息")

public class RabbitProduct {

    @Resource

    private RabbitTemplate rabbitTemplate;

    /**

     * 构造方法注入rabbitTemplate

     */

    @Autowired

    public RabbitProduct(RabbitTemplate rabbitTemplate){

        this.rabbitTemplate = rabbitTemplate;

    }

    @Resource

    private Redis redis;

    //发送消息 推送到websocket    参数自定义 转为String发送消息

    @GetMapping("/mm")

    @ApiOperation(value = "发消息")

    public String sendMSG(String msg,ConfigurationTable configurationTable){

        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());

        redis.selDateKey(1,"ConfigurationType",configurationTable.getConfigurationType(),500);

        //把消息对象放入路由对应的队列当中去

        rabbitTemplate.convertAndSend(configurationTable.getRoute(),configurationTable.getBindingKey(), JSONObject.toJSON(msg).toString(), correlationId);

        return "操作成功！";

    }

    public String sendMsg(String msg,ConfigurationTable configurationTable){

        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());

        //把消息对象放入路由对应的队列当中去

        rabbitTemplate.convertAndSend(configurationTable.getRoute(),configurationTable.getBindingKey(), JSONObject.toJSON(msg).toString(), correlationId);

        return "操作成功！";

    }

}

接下来就是接收消息了，我这里使用的是手动ACK的方式，想要简单一点的话，用注解直接监听，把rabbitConfig中的手动ack删了就好

package com.example.demotest.listener;

import com.example.demotest.config.WebSocketServerEndpoint;

import com.example.demotest.util.Redis;

import com.rabbitmq.client.Channel;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.annotation.Exchange;

import org.springframework.amqp.rabbit.annotation.Queue;

import org.springframework.amqp.rabbit.annotation.QueueBinding;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

import javax.annotation.Resource;

import java.io.IOException;

/**

 * Description :

 * Copyright : Copyright (c) 2022

 * Company : 中移（成都）产业研究院 Co. Ltd.

 *

 * @author daisuhang

 * @version 1.0.0

 * @createTime 2022年11月14日 09:07:00

 */

@Slf4j

@Component

public class RabbitNewConsumer {

    @Autowired

    private Redis redis;

    private static RabbitNewConsumer rabbitNewConsumer;

    @Resource

    private WebSocketServerEndpoint webSocketServerEndpoint; //引入WebSocket

    /**

     * 构造方法注入rabbitTemplate

     */

    @PostConstruct

    public void init() {

        rabbitNewConsumer = this;

        rabbitNewConsumer.webSocketServerEndpoint = webSocketServerEndpoint;

    }

    @RabbitListener(bindings = @QueueBinding(

                      value = @Queue(value = "new_queue",durable = "true"),

                     exchange = @Exchange(name = "new_exchang",durable = "true",type = "direct"),

                     key = "me_key"

                     )) //监听队列

    public void msgReceive(String content, Message message, Channel channel) throws IOException {

        log.info("----------------接收到消息--new-queue------------------"+content);

        //发送给WebSocket 由WebSocket推送给前端

        String configurationType = redis.get("ConfigurationType");

        if (configurationType.equals("事件上传")){

            rabbitNewConsumer.webSocketServerEndpoint.sendMessageOnline(content);

        }

        // 确认消息已接收

        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

    }

}

下面这个是加了休眠时间的延时

package com.example.demotest.listener;

import com.example.demotest.config.RabbitConfig;

import com.example.demotest.config.WebSocketServerEndpoint;

import com.example.demotest.util.Redis;

import com.rabbitmq.client.Channel;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.annotation.Exchange;

import org.springframework.amqp.rabbit.annotation.Queue;

import org.springframework.amqp.rabbit.annotation.QueueBinding;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

import javax.annotation.Resource;

import java.io.IOException;

import static com.example.demotest.config.RabbitConfig.*;

/**

 * @title

 * @description 定时消息队列 消费监听回调

 * @author admin  issuser

 * @updateTime 2022/11/10 15:43

 * @throws

 */

@Slf4j

@Component

public class RabbitConsumer {

    private static RabbitConsumer rabbitConsumer;

    @Resource

    private WebSocketServerEndpoint webSocketServerEndpoint; //引入WebSocket

    @Autowired

    private Redis redis;

    /**

     * 构造方法注入rabbitTemplate

     */

    @PostConstruct

    public void init() {

        rabbitConsumer = this;

        rabbitConsumer.webSocketServerEndpoint = webSocketServerEndpoint;

    }

    @RabbitListener(bindings = @QueueBinding(

            value = @Queue(value = msg_queue,durable = "true"),

            exchange = @Exchange(name = msg_exchang,durable = "true",type = "direct"),

            key = msg_routing_key

    )) //监听队列

    public void msgReceive(String content, Message message, Channel channel) throws IOException {

        log.info("----------------接收到消息-msg_queue-------------------"+content);

        //发送给WebSocket 由WebSocket推送给前端

        String configurationType = redis.get("ConfigurationType");

        if (configurationType.equals("属性上传")){

            rabbitConsumer.webSocketServerEndpoint.sendMessageOnline(content);

            // 确认消息已接收

            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

        }

        try {

            Thread.sleep(2000);

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);

    }

}

接下来就是socket的一些配置

package com.example.demotest.config;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.web.socket.server.standard.ServerEndpointExporter;

/**

 * WebSocket配置类

 */

@Configuration

public class WebSocketConfig {

    @Bean

    public ServerEndpointExporter serverEndpointExporter() {

        return new ServerEndpointExporter();

    }

}

连接前端的配置，以及一些方法封装

package com.example.demotest.config;

import lombok.extern.slf4j.Slf4j;

import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

import javax.websocket.*;

import javax.websocket.server.PathParam;

import javax.websocket.server.ServerEndpoint;

import java.io.IOException;

import java.util.concurrent.ConcurrentHashMap;

/**

 *  WebSocket 服务配置类

 *  定义 userId 为当前连接(在线) WebSocket 的用户

 */

@Slf4j

@Component

@ServerEndpoint(value = "/ws/{userId}")

public class WebSocketServerEndpoint {

    private Session session; //建立连接的会话

    private String userId; //当前连接用户id   路径参数

    /**

     * 存放存活的Session集合(map保存)

     */

    private static ConcurrentHashMap<String , WebSocketServerEndpoint> livingSession = new ConcurrentHashMap<>();

    /**

     *  建立连接的回调

     *  session 建立连接的会话

     *  userId 当前连接用户id   路径参数

     */

    @OnOpen

    public void onOpen(Session session,  @PathParam("userId") String userId){

        this.session = session;

        this.userId = userId;

        livingSession.put(userId, this);

        log.debug("----[ WebSocket ]---- 用户id为 : {} 的用户进入WebSocket连接 ! 当前在线人数为 : {} 人 !--------",userId,livingSession.size());

    }

    /**

     *  关闭连接的回调

     *  移除用户在线状态

     */

    @OnClose

    public void onClose(){

        livingSession.remove(userId);

        log.debug("----[ WebSocket ]---- 用户id为 : {} 的用户退出WebSocket连接 ! 当前在线人数为 : {} 人 !--------",userId,livingSession.size());

    }

    @OnMessage

    public void onMessage(String message, Session session,  @PathParam("userId") String userId) {

        log.debug("--------收到用户id为 : {} 的用户发送的消息 ! 消息内容为 : ------------------",userId,message);

        //sendMessageToAll(userId + " : " + message);

    }

    @OnError

    public void onError(Session session, Throwable error) {

        log.error("----------------WebSocket发生错误----------------");

        log.error(error.getStackTrace() + "");

    }

    /**

     *  根据userId发送给用户

     * @param userId

     * @param message

     */

    public void sendMessageById(String userId, String message) {

        livingSession.forEach((sessionId, session) -> {

            //发给指定的接收用户

            if (userId.equals(session.userId)) {

                sendMessageBySession(session.session, message);

            }

        });

    }

    /**

     *  根据Session发送消息给用户

     * @param session

     * @param message

     */

    public void sendMessageBySession(Session session, String message) {

        try {

            session.getBasicRemote().sendText(message);

        } catch (IOException e) {

            log.error("----[ WebSocket ]------给用户发送消息失败---------");

            e.printStackTrace();

        }

    }

    public void sendMessageBySessionObject(Session session, Object message) {

        try {

            session.getBasicRemote().sendObject(message);

        } catch (IOException | EncodeException e) {

            log.error("----[ WebSocket ]------给用户发送消息失败---------");

            e.printStackTrace();

        }

    }

    /**

     *  给在线用户发送消息

     * @param message

     */

    public void sendMessageOnline(String message) {

        livingSession.forEach((sessionId, session) -> {

            if(session.session.isOpen()){

                sendMessageBySession(session.session, message);

            }

        });

    }

    /**

     *  给在线用户发送消息

     * @param message

     */

    public void sendMessageOnlineObject(Object message) {

        livingSession.forEach((sessionId, session) -> {

            if(session.session.isOpen()){

                sendMessageBySessionObject(session.session, message);

            }

        });

    }

}

最后就是前端页面了

<!DOCTYPE HTML>

<html>

<head>

    <title>My WebSocket</title>

</head>

<body>

Welcome<br/>

<input id="text" type="text" /><button onclick="send()">Send</button>    <button onclick="closeWebSocket()">Close</button>

<div id="message">

</div>

</body>

<script type="text/javascript">

    var websocket = null;

    //判断当前浏览器是否支持WebSocket

    if('WebSocket' in window){

        websocket = new WebSocket("ws://localhost:8001/ws/{123}");

    }

    else{

        alert('当前浏览器 Not support websocket')

    }

    //连接发生错误的回调方法

    websocket.onerror = function(){

        setMessageInnerHTML("error");

    };

    //连接成功建立的回调方法

    websocket.onopen = function(event){

        setMessageInnerHTML("WebSocket连接成功");

    }

    //接收到消息的回调方法

    websocket.onmessage = function(event){

        setMessageInnerHTML(event.data);

    }

    //连接关闭的回调方法

    websocket.onclose = function(){

        setMessageInnerHTML("WebSocket连接关闭");

    }

    //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。

    window.onbeforeunload = function(){

        websocket.close();

    }

    //将消息显示在网页上

    function setMessageInnerHTML(innerHTML){

        document.getElementById('message').innerHTML += innerHTML + '<br/>';

    }

    //关闭连接

    function closeWebSocket(){

        websocket.close();

    }

    //发送消息

    function send(){

        var message = document.getElementById('text').value;

        websocket.send(message);

    }

</script>

</html>

到此本文就结束了

巴特西

java springboot+rabbitmq+websocket 订阅展示

最新文章

热门文章