如何安裝rabbitmq在此就不再贅述了，直接上程式碼，使用的direct佇列模式。

依賴

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>
 

生產者

application.properties

#rabbitmq配置
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
#開啟訊息傳送確認
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
#返回無法插入佇列的訊息
spring.rabbitmq.template.mandatory=true

準備訊息佇列與路由key

public class RabbitConstant {
    //交換機名稱
    public final static String EXCHANGE = "exchange_test";
    //佇列
    public final static String QUEUE_TRANSACTION = "queue_transaction";
    public final static String QUEUE_CONTRACT = "queue_contract";
    public final static String QUEUE_QUALIFICATION = "queue_qualification";
    //路由key
    public final static String RK_TRANSACTION = "transaction";
    public final static String RK_CONTRACT = "contract";
    public final static String RK_QUALIFICATION = "qualification";
}
 

配置

@Configuration
public class RabbitMqConfig {

    /**
     * 宣告佇列
     *
     * @return
     */
    @Bean
    public Queue queueTransaction() {
        // true表示持久化該佇列
        return new Queue(RabbitConstant.QUEUE_TRANSACTION, true);
    }

    @Bean
    public Queue queueContract() {
        // true表示持久化該佇列
        return new Queue(RabbitConstant.QUEUE_CONTRACT, true);
    }

    @Bean
    public Queue queueQualification() {
        // true表示持久化該佇列
        return new Queue(RabbitConstant.QUEUE_QUALIFICATION, true);
    }

    /**
     * 宣告互動器
     *
     * @return
     */
    @Bean
    DirectExchange directExchange() {
        return new DirectExchange(RabbitConstant.EXCHANGE);
    }

    /**
     * 繫結
     *
     * @return
     */
    @Bean
    public Binding bindingTransaction() {
        return BindingBuilder.bind(queueTransaction()).to(directExchange()).with(RabbitConstant.RK_TRANSACTION);
    }

    /**
     * 繫結
     *
     * @return
     */
    @Bean
    public Binding bindingContract() {
        return BindingBuilder.bind(queueContract()).to(directExchange()).with(RabbitConstant.RK_CONTRACT);
    }

    /**
     * 繫結
     *
     * @return
     */
    @Bean
    public Binding bindingQualification() {
        return BindingBuilder.bind(queueQualification()).to(directExchange()).with(RabbitConstant.RK_QUALIFICATION);
    }

}

釋出者實現介面

@Component
@Slf4j
public class Sender implements RabbitTemplate.ConfirmCallback, ReturnCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    //訊息傳送確認回撥方法
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("訊息傳送成功:" + correlationData);
    }

    //訊息傳送失敗回撥方法
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("訊息傳送失敗:" + new String(message.getBody()));
    }

    /**
     * 傳送訊息，不需要實現任何介面，供外部呼叫
     *
     * @param messageVo
     */
    public void send(MessageVo messageVo) {
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(RabbitConstant.EXCHANGE, messageVo.getRoutingKey(), JSONObject.toJSON(messageVo).toString(), correlationId);
    }
}

Ok,只要呼叫send方法就可一把訊息傳送到指定的佇列中了，傳送後會呼叫confirm或者returnedMessage方法來返回結果，

下面開始講消費者。

消費者

application.properties

#rabbitmq配置
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

消費者配置

import com.rabbitmq.client.Channel;
import com.yxd.rabbitmq.web.websocket.WebSocketServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;

import java.io.IOException;


@Configuration
@EnableRabbit
@Slf4j
public class ConsumerConfig implements RabbitListenerConfigurer {

    @Autowired
    private ConnectionFactory connectionFactory;

    @Autowired
    private WebSocketServerEndpoint webSocketServerEndpoint;

    @Bean
    public DefaultMessageHandlerMethodFactory handlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setMessageConverter(new MappingJackson2MessageConverter());
        return factory;
    }

    @Bean
    public SimpleMessageListenerContainer messageContainer(@Qualifier("queueTransaction") Queue transaction, @Qualifier("queueContract") Queue contract, @Qualifier("queueQualification") Queue qualification) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(transaction, contract, qualification);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //設定確認模式手工確認
        container.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                byte[] body = message.getBody();
                log.info("receive msg : " + new String(body));
                try {
                    webSocketServerEndpoint.sendMessageToAll(new String(body));
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//確認訊息成功消費
                } catch (IOException e) {
                    log.error("訊息推送前臺出錯：" + e.getMessage() + "/r/n重新發送");
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); //重新發送
                }
            }
        });
        return container;
    }

    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar rabbitListenerEndpointRegistrar) {
        rabbitListenerEndpointRegistrar.setMessageHandlerMethodFactory(handlerMethodFactory());
    }
}

配置完後會自動監聽訊息佇列，AcknowledgeMode設定為手動模式可以自己來管理是否確認消費成功或者重新發送，當然我這裡重新發送會不斷的傳送，到時候根據具體業務進行處理就行了。

websocket

配置

@Configuration
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

}

socket服務

package com.yxd.rabbitmq.web.websocket;

import com.alibaba.fastjson.JSONObject;
import com.yxd.rabbitmq.model.MessageVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ServerEndpoint
 * <p>
 * 使用springboot的唯一區別是要@Component宣告下，而使用獨立容器是由容器自己管理websocket的，但在springboot中連容器都是spring管理的。
 * <p>
 * 雖然@Component預設是單例模式的，但springboot還是會為每個websocket連線初始化一個bean，所以可以用一個靜態set儲存起來。
 */
@ServerEndpoint("/ws/yxd/{userId}") //WebSocket客戶端建立連線的地址
@Component
@Slf4j
public class WebSocketServerEndpoint {

    /**
     * 存活的session集合（使用執行緒安全的map儲存）
     */
    private static Map<String, Session> livingSessions = new ConcurrentHashMap<>();

    /**
     * 建立連線的回撥方法
     *
     * @param session 與客戶端的WebSocket連線會話
     * @param userId  使用者名稱，WebSocket支援路徑引數
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        livingSessions.put(session.getId(), session);
        log.info(userId + "進入連線");
    }

    @OnMessage
    public void onMessage(String message, Session session, @PathParam("userId") String userId) {
        log.info(userId + " : " + message);
        sendMessageToAll(userId + " : " + message);
    }


    @OnError
    public void onError(Session session, Throwable error) {
        log.info("發生錯誤");
        log.error(error.getStackTrace() + "");
    }


    @OnClose
    public void onClose(Session session, @PathParam("userId") String userId) {
        livingSessions.remove(session.getId());
        log.info(userId + " 關閉連線");
    }

    /**
     * 單獨傳送訊息
     *
     * @param session
     * @param message
     */
    public void sendMessage(Session session, String message) {
        try {
            session.getBasicRemote().sendText(message);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 群發訊息
     *
     * @param message
     */
    public void sendMessageToAll(String message) {
        MessageVo messageVo = JSONObject.parseObject(message, MessageVo.class);
        livingSessions.forEach((sessionId, session) -> {
            //發給指定的接收使用者
//            if (userId.equals(messageVo.getReceiveUserId())) {
            sendMessage(session, message);
//            }
        });
    }

}

前臺

我這裡使用的是vue

queueReceiveSetting: {//訊息佇列配置
          websock: null,
          client: null,
          wsuri: "ws://localhost:8111/rabbitmq/ws/huhy/lxy",
        },

initWebSocket() {
        //ws地址
        if (this.queueReceiveSetting.websock) {
          this.queueReceiveSetting.websock.close()
        }
        this.queueReceiveSetting.websock = new WebSocket(this.queueReceiveSetting.wsuri);
        this.queueReceiveSetting.websock.onopen = function (res) {
          console.log("開啟連線")
        };
        this.queueReceiveSetting.websock.onmessage = function (res) {
          let data = JSON.parse(res.data)
          console.log("收到資料：" + data.content)
          Message({
            message: data.content,
            type: 'info',
            showClose: true,
            duration: 3 * 1000
          })
        };
        this.queueReceiveSetting.websock.onclose = function (res) {
          console.log("連線關閉")
        };
        this.queueReceiveSetting.websock.onerror = function (res) {
          console.log("連接出錯")
          // this.initWebSocket();
        };
      }