欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

SpringBoot2實現(xiàn)MessageQueue消息隊列

 更新時間：2023年04月25日 09:47:34   作者：yjx23332  
本文主要介紹了 SpringBoot2實現(xiàn)MessageQueue消息隊列,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

什么是消息隊列

通常說的消息隊列，簡稱MQ（Message Queue），指的就是消息中間件。簡單理解為一個使用隊列來通信的組件，本質上就是個轉發(fā)器，包含發(fā)消息，存消息，消費消息的過程。

一、異步與同步

1.1 同步通訊與異步通訊

  • 同步通訊：時效性強。比如視頻電話，實時傳到對方，同時對方出回應。
  • 異步通訊：比如網絡聊天，非實時反饋的，不會立即得到結果，可以之后再回復。

1.2 同步調用的問題

微服務基于Feign的調用就是同步的方式。

以購物場景為例

但是如果要加業(yè)務就需要為支付服務加業(yè)務，改動其代碼，耦合度高。

同時，同步調用，要等待服務結束后，在進行下一個服務。支付總耗時，是支付服務依次調用服務耗時的時間和，耗時過長。

此外，如果倉儲服務掛掉了，支付服務就會被卡在那里。當過多的支付服務都卡在那里，于是資源耗盡，支付服務也掛掉了。

問題：

  • 耦合度高
  • 性能下降
  • 資源浪費
  • 級聯(lián)失敗

1.3 異步調用方案

異步調用常見的就是事件驅動模式

當支付服務告知了Broker后，就可以繼續(xù)自己的事情了，而不需要等待。

優(yōu)勢：

  • 代碼解耦合：不需要改動支付服務，只需要讓服務訂閱或者取消訂閱Broker即可。
  • 耗時減少了：只計算支付服務和通知Broker的時間。
  • 不存在級聯(lián)失敗的問題，倉儲服務掛了不再影響支付服務。
  • 流量消峰：當流量過大時，請求排在Broker中，服務能做幾個就做幾個，做不了的就排著。

缺點：

  • Broker掛了也會出問題，依賴于Broker的可靠性，安全性，吞吐能力
  • 架構復雜了，業(yè)務沒有明顯的流程線，不好追蹤管理

二、MQ消息隊列

在上述的結構中，就是Broker。

常用的MQ有幾種實現(xiàn)。

RabbitMQActiveMQRocketMQKafaka
公司/社區(qū)RabbitApache阿里Apache
開發(fā)語言ErlangJavaJavaScala&Java
協(xié)議支持AMQP、XMPP、SMTP、STOMPOpenWire、STOMP、REST、XMPP、AMQP自定義協(xié)議自定義協(xié)議
可用性高一般高高
單機吞吐量一般差高非常高
消息延遲微秒級毫秒級毫秒級毫秒以內
消息可靠性高一般高一般
  • 一般中小型公司，用的就是RabbitMQ。
  • 如果大型企業(yè)，做深度定制，可以用RocketMQ
  • Kafaka則是用于大量數據情況下的處理，但安全可靠性相對較差。
  • ActiveMQ是很早的消息隊列，如今幾乎沒有維護。

2.1 單機部署MQ

通過docker部署最簡單，

docker pull rabbitmq:3-management

也可以用命令安裝,這里直接用容器了。

啟動信息如下

docker run -e RABBITMQ_DEFAULT_USER=yjx23332 -e RABBITMQ_DEFAULT_PASS=123456 -v mq-plugins:/plugins --name mq --hostname mq1 -p 15672:15672 -p 5672:5672 -d rabbitmq:3-management

-e 為設置環(huán)境變量
兩個端口，15672是管理平臺端口，5672是發(fā)送消息的端口。

記得開放對應端口。如果是騰訊云或者阿里云，也要在購買的服務器管理頁面打開放行端口。

firewall-cmd --zone=public --add-port=15672/tcp --permanent
firewall-cmd --zone=public --add-port=5672/tcp --permanent
firewall-cmd --reload
查詢端口是否開放

firewall-cmd --query-port=15672/tcp 查看某個端口firewall-cmd --zone=public --list-ports 查看所有

登陸成功后，即可進入以下界面。

我們可在這里為添加用戶和角色

virtualhosts虛擬主機：對不用戶進行隔離，避免相互影響。
此處可以添加

點擊用戶，可以配置其虛擬主機權限等。

此處設置交換機

2.2 結構和概念

使用消息隊列中消息的對象。我們稱之為消費者。

在一個virtualhost下：

2.3 常見的消息模型

基本消息隊列BasicQueue：最簡單的實現(xiàn)

工作消息隊列WorkQueue：在工作者之間分配任務

發(fā)布訂閱帶有交換機，分為：

Fanout Exchange：廣播，發(fā)布訂閱（publish/subscribe）：一次性向讀個消費者發(fā)送消息。

Direct Exchange：路由（Routing）：有選擇的接收消息

Topic Exchange：主題 （Topics）：根據主題接收消息。

請求回復模型（RPC）：收到請求然后答復。

發(fā)布者確認模式（Publisher Confirms）：會讓發(fā)布者知道發(fā)送是否成功。

三、SpringAMQP

3.1 用非自動裝配的方式使用消息隊列

需要在項目中引入AMQP，記得加入父類spring-boot-starter-parent

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

引入Junit方便測試

		<dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <scope>test</scope>
        </dependency>

測試一下MQ

我們創(chuàng)建如下兩個子項目。

為test寫一個測試用例

package com.yjx23332.mq.helloworld;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;



public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException{
        //1.建立連接
        ConnectionFactory factory = new ConnectionFactory();
        //1.2.設置連接參數
        factory.setHost("IP");//MQ地址設置
        factory.setPort(5672);//端口設置
        factory.setVirtualHost("/");//設置虛擬主機
        factory.setUsername("賬號");
        factory.setPassword("密碼");
        //1.2 建立連接
        Connection connection = factory.newConnection();

        //2.創(chuàng)建通道
        Channel channel = connection.createChannel();

        //3.創(chuàng)建消息隊列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName,false,false,false,null);

        //4.發(fā)送消息
        String message = "hello,rabbitmq!";
        channel.basicPublish("",queueName,null,message.getBytes());
        System.out.println("已發(fā)送消息：【"+message+"】");

        //5.關閉通道
        if(channel != null){
            channel.close();
        }
        if(connection != null){
            connection.close();
        }
    }
}

我們在發(fā)送消息前打上斷點，用junit運行，就可以看到連接創(chuàng)建和通道創(chuàng)建.。

因為發(fā)完就不管了，因此必須打斷點，才看得到連接和通道。

完成后可以看到隊列中

接下來我們處理消息,基本一樣，只需要修改幾個部分。

注意我們沒有關閉連接，因為在業(yè)務中，要一直處理。

package com.yjx23332.mq.helloworld;

import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


public class ConsumerTest {
    @Test
    public static void main(String[] args) throws IOException, TimeoutException{
        //1.建立連接
        ConnectionFactory factory = new ConnectionFactory();
        //1.2.設置連接參數
        factory.setHost("101.43.65.53");//MQ地址設置
        factory.setPort(5672);//端口設置
        factory.setVirtualHost("/");//設置虛擬主機
        factory.setUsername("yjx23332");
        factory.setPassword("123456");
        //1.2 建立連接
        Connection connection = factory.newConnection();

        //2.創(chuàng)建通道
        Channel channel = connection.createChannel();

        //3.創(chuàng)建消息隊列
        //為什么這里也要創(chuàng)建？避免消費者先執(zhí)行，還沒有隊列。同時，相同的隊列創(chuàng)建重復執(zhí)行沒有影響。
        String queueName = "simple.queue";
        channel.queueDeclare(queueName,false,false,false,null);

        //4.處理消息
        String message = "hello,rabbitmq!";
        //DefaultConsumer 是回調函數，一旦有消息，異步處理
        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws  IOException{
                System.out.println("接收到消息：【"+ new String (body)+"】");
            }
        });
        System.out.println("####################等待接收消息##################");
//
//        //5.關閉通道
//        if(channel != null){
//            channel.close();
//        }
//        if(connection != null){
//            connection.close();
//        }
    }
}

結果如下

可以看到，因為回調的原因，后面的輸出先執(zhí)行

隊列中消息處理完畢

3.2 SpringAMQP介紹

AMQP：Advanced Message Queuing Protocol：高級消息隊列協(xié)議。于應用程序之間傳遞業(yè)務消息的開放標準。

Spring AMQP：基于AMQP協(xié)議的一套API規(guī)范，提供模板來發(fā)送和接收消息。其中Spring-amqp是基礎抽象，Spring-rabbit是底層的默認實現(xiàn)?？蓞⒖?a rel="external nofollow" rel="external nofollow" target="_blank">Spring AMQP官網。

3.3 基礎消息隊列功能使用

導入依賴

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

為了方便測試，我們引入SpringBoot單元測試

	<dependency>
  		<groupId>org.springframework.boot</groupId>
  		<artifactId>spring-boot-starter-test</artifactId>
  		<scope>test</scope>
	</dependency>
	<dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
    	<scope>test</scope>
    </dependency>

然后準備一個yml文件，配置和之前用代碼寫得相似。

spring:
  rabbitmq:
    host: 
    port: 5672
    virtual-host: /
    username: 
    password:

我們直接走單元測試，這里就不創(chuàng)建一個隊列了，直接放消息。

package com.yjx23332.mq.helloworld;


import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


@SpringBootTest
@RunWith(SpringRunner.class)
public class PublisherTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSimpleQueue() throws IOException, TimeoutException{
        rabbitTemplate.convertAndSend("simple.queue","hello,spring amqp");
    }
}

接下來為消費者建一個監(jiān)聽器(記得配置yml文件)

package com.yjx23332.mq.listener;

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

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException{
        System.out.println("spring 消費者接收到消息:【"+msg+"】");
    }
}

消息一旦消費，就會被移除，Rabbit MQ不存在回溯功能。

3.4 工作隊列的配置

一個隊列綁定多個消費者。

我們準備發(fā)送50條消息

package com.yjx23332.mq.helloworld;


import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;



@SpringBootTest
@RunWith(SpringRunner.class)
public class PublisherTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSimpleQueue() throws InterruptedException{
        for(int i = 0;i < 50;i++){
            rabbitTemplate.convertAndSend("simple.queue","hello,spring amqp___" + i);
            Thread.sleep(20);
        }
    }
}

修改消費者

package com.yjx23332.mq.listener;

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

import java.time.LocalTime;

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException{
        System.out.println("spring 消費者接收到消息:【"+msg+"】" + LocalTime.now());
        Thread.sleep(20);
    }
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage2(String msg) throws InterruptedException{
        System.err.println("spring 消費者接收到消息:【"+msg+"】"+ LocalTime.now());
        Thread.sleep(200);
    }
}

從結果會發(fā)現(xiàn)處理總時長超過了1秒達到了5秒，查看輸出會發(fā)現(xiàn)消息被平均分配給了兩個。一個處理偶數，一個處理奇數。但由于處理速度不同，因此處理總時長超過了1秒。

這里是因為消費預取導致的，在執(zhí)行前會提前把消息從隊列拿出，然后各自處理。

但我們希望的是，做的快的多做，做的慢的少做。

因此我們可以修改yml文件：

spring:
  rabbitmq:
    host: 
    port: 5672
    virtual-host: /
    username: 
    password: 
    listener:
      simple:
        prefetch:  1 # 每次只能獲取幾條消息，執(zhí)行完了再取下一條，默認是無限

重啟后再次執(zhí)行就會發(fā)現(xiàn)正常了。

3.5 發(fā)布與訂閱模式

我們需要將同一消息發(fā)送給多個消費者。需要加入交換機來實現(xiàn)。注意，交換機只負責消費路由，但不存儲消息，丟失一概不負責。

3.5.1 SpringAMQP交換機類

3.5.2 Fanout Exchange

我們在consumer服務中聲明Exchange、Queue、Binding.

package com.yjx23332.mq.confg;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutConfig {
    //聲明FanoutExchange交換機
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("yjx23332.fanout");
    }
    //聲明一個隊列
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
    //綁定隊列和交換機
    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
    //聲明第二個隊列
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
    //綁定第二個隊列和交換機
    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

運行后，會看到：

修改監(jiān)聽器

package com.yjx23332.mq.listener;

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


@Component
public class MQlistener {
    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg){
        System.out.println("spring 消費者接收q1到消息:【"+msg+"】");
    }
    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg){
        System.err.println("spring 消費者接收到q2消息:【"+msg+"】");
    }
}

我們再修改publisher的測試代碼

package com.yjx23332.mq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@SpringBootTest
@RunWith(SpringRunner.class)
public class test {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testFanoutExchange(){
        String exchangeName = "yjx23332.fanout";
        String message = "hello world!";
        rabbitTemplate.convertAndSend(exchangeName,"",message);
    }
}

啟動

3.5.3 DirectExchange

將接收到的消息根據規(guī)則路由到指定的Queue，因此稱為路由模式（routes）。

  • 每一個Queue都與Exchange設置一個BindingKey
  • 發(fā)布者發(fā)送消息時，指定消息的RoutingKey
  • Exchange將消息路由到BindingKey與消息RoutingKey一致的隊列
  • 一個隊列可以指定多個BindingKey，且隊列之間的BindingKey可以重復

由于基于Config創(chuàng)建隊列交換機的方式很麻煩，我們用新的方式聲明交換機、隊列。

刪除上一節(jié)我們在config中的聲明代碼。

然后在listener中進行

package com.yjx23332.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
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.stereotype.Component;


@Component
public class MQlistener {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(value = "yjx23332.direct",type = ExchangeTypes.DIRECT),
            key = {"red","blue"} //bindingkey
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("spring 消費者接收q1到消息:【"+msg+"】");
    }
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(value = "yjx23332.direct" , type = ExchangeTypes.DIRECT),
            key = {"red","yellow"}
    ))
    public void listenDirectQueue2(String msg) {
        System.err.println("spring 消費者接收到q2消息:【"+msg+"】");
    }
}

運行后，我們可以看到

接下來，我們修改Test代碼

package com.yjx23332.mq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@SpringBootTest
@RunWith(SpringRunner.class)
public class test {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testDirectExchange(){
        String exchangeName = "yjx23332.direct";
        rabbitTemplate.convertAndSend(exchangeName,"red","hello red");
        rabbitTemplate.convertAndSend(exchangeName,"blue","hello blue");
        rabbitTemplate.convertAndSend(exchangeName,"yellow","hello yellow");
    }
}

3.5.4 TopicExchange

與DirectExchange類似，但是它的routingKey必須是多個單詞表，并用’.'分割。
當隊列與交換機綁定時，可以使用通配符。避免當bindkey過多導致的麻煩。

#：代表0個或多個單詞
*：代指一個單詞

比如

China.news
Japan.news
就可以用 #.news
同理
China.weather
China.news
就可以用 China.#

我們沿用上一節(jié)的代碼,做一點修改即可

package com.yjx23332.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
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.stereotype.Component;


@Component
public class MQlistener {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue"),
            exchange = @Exchange(value = "yjx23332.topic",type = ExchangeTypes.TOPIC),
            key = {"China.#"} //bindingkey
    ))
    public void listenTopicQueue1(String msg){
        System.out.println("spring 消費者接收q1到消息:【"+msg+"】");
    }
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(value = "yjx23332.topic",type = ExchangeTypes.TOPIC),
            key = {"#.news"}
    ))
    public void listenTopicQueue2(String msg) {
        System.err.println("spring 消費者接收到q2消息:【"+msg+"】");
    }
}

重啟后，可看到

修改Test代碼

package com.yjx23332.mq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@SpringBootTest
@RunWith(SpringRunner.class)
public class test {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testDirectExchange(){
        String exchangeName = "yjx23332.topic";
        rabbitTemplate.convertAndSend(exchangeName,"China.news","江蘇地表最高溫度將達到72攝氏度");
        rabbitTemplate.convertAndSend(exchangeName,"China.weather","未來溫度仍將升高");
        rabbitTemplate.convertAndSend(exchangeName,"Japan.news","安培中槍");
    }
}

3.6 消息轉換器

在發(fā)送中，我們接收消息的類型是Object。SpringAMQP會幫我們序列化后變?yōu)樽止?jié)發(fā)送。
用默認JDK的序列化ObjectOutputStream是沒有問題的，但是中間過程是亂碼，我們這里改用JSON方式的序列化，這樣在消息隊列中查看也是正常的。

默認JDK的消息信息：

接下來我們配置消息轉換。

我們先在消費者聲明一個queue，并設置處理方式

package com.yjx23332.mq.listener;


import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;


@Component
public class MQlistener {
    @RabbitListener(queuesToDeclare = @Queue("object.queue"))
    public void listenObjectQueue(String msg){
        System.out.println("spring 消費者接收到Object消息:【"+msg+"】");
    }
}

我們?yōu)榘l(fā)送類引入依賴并編寫配置

		<dependency>
            <groupId>com.fasterxml.jackson.dataformat</groupId>
            <artifactId>jackson-dataformat-yaml</artifactId>
        </dependency>

覆蓋默認的消息轉換。

package com.yjx23332.mq.config;


import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MessageConverterConfig {
    @Bean
    public MessageConverter jsonMessageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

隨后修改Test

package com.yjx23332.mq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;

@SpringBootTest
@RunWith(SpringRunner.class)
public class test {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testObjectQueue(){
        String queueName = "object.queue";
        Map<String,Object> msg = new HashMap<>();
        msg.put("name","yjx23332");
        msg.put("age",21);
        rabbitTemplate.convertAndSend(queueName,msg);
    }
}

結果如下：

這時消息不再是亂碼

我們在為消費者配置轉換，并修改監(jiān)聽器。當然，如果我們在兩邊都不配置消息轉換器，這里結果是一樣的。

package com.yjx23332.mq.listener;


import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;


@Component
public class MQlistener {
    @RabbitListener(queuesToDeclare = @Queue("object.queue"))
    public void listenObjectQueue(Map<String,Object> msg){
        System.out.println("spring 消費者接收到Object消息:【 name = "+msg.get("name")+",age = "+msg.get("age")+"】");
    }

}

結果如下

參考文獻

[1]Spring AMQP官網
[2]黑馬程序員Java微服務
[3]RabbitMQ官方文檔

到此這篇關于 SpringBoot2實現(xiàn)MessageQueue消息隊列的文章就介紹到這了,更多相關 SpringBoot2 MessageQueue消息隊列內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

相關文章

  • Spring Security源碼解析之權限訪問控制是如何做到的

    Spring Security源碼解析之權限訪問控制是如何做到的

    Spring Security 中對于權限控制默認已經提供了很多了，但是，一個優(yōu)秀的框架必須具備良好的擴展性，下面小編給大家介紹Spring Security源碼解析之權限訪問控制是如何做到的，感興趣的朋友跟隨小編一起看看吧
    2021-05-05
  • Struts2學習筆記(2)-路徑問題解決

    Struts2學習筆記(2)-路徑問題解決

    本文主要介紹Struts2的路徑問題，盡量不要使用相對路徑，使用相對路徑會讓路徑問題變得很繁瑣很麻煩，推薦使用絕對路徑，希望能給大家做一個參考。
    2016-06-06
  • java實現(xiàn)停車場管理系統(tǒng)

    java實現(xiàn)停車場管理系統(tǒng)

    這篇文章主要為大家詳細介紹了java實現(xiàn)停車場管理系統(tǒng)，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
    2019-11-11
  • 解決SpringCloud Config結合github無法讀取配置的問題

    解決SpringCloud Config結合github無法讀取配置的問題

    這篇文章主要介紹了解決SpringCloud Config結合github無法讀取配置的問題，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
    2021-02-02
  • 詳解@Autowired(required=false)注入注意的問題

    詳解@Autowired(required=false)注入注意的問題

    這篇文章主要介紹了@Autowired(required=false)注入注意的問題,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
    2020-04-04
  • 基于JavaSwing+mysql開發(fā)一個學生社團管理系統(tǒng)設計和實現(xiàn)

    基于JavaSwing+mysql開發(fā)一個學生社團管理系統(tǒng)設計和實現(xiàn)

    項目使用Java swing+mysql開發(fā)，可實現(xiàn)基礎數據維護、用戶登錄注冊、社團信息列表查看、社團信息添加、社團信息修改、社團信息刪除以及退出注銷等功能、界面設計比較簡單易學、適合作為Java課設設計以及學習技術使用，需要的朋友參考下吧
    2021-08-08
  • MyBatis Generator 自定義生成注釋的方法

    MyBatis Generator 自定義生成注釋的方法

    這篇文章主要介紹了MyBatis Generator 自定義生成注釋的方法，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧
    2018-09-09
  • Spring Security使用數據庫認證及用戶密碼加密和解密功能

    Spring Security使用數據庫認證及用戶密碼加密和解密功能

    這篇文章主要介紹了Spring Security使用數據庫認證及用戶密碼加密和解密,本文通過代碼與截圖的形式給大家介紹的非常詳細，對大家的工作或學習具有一定的參考借鑒價值，需要的朋友可以參考下
    2020-03-03
  • Java中占位符的超全使用方法分享

    Java中占位符的超全使用方法分享

    這篇文章主要為大家詳細介紹了Java中常見的一些占位符的使用方法，例如%d，%s等，文中的示例代碼簡潔易懂，感興趣的小伙伴可以跟隨小編一起學習學習
    2023-05-05
  • Spring一步到位精通攔截器

    Spring一步到位精通攔截器

    攔截器（Interceptor）是一種動態(tài)攔截方法調用的機制，在SpringMVC中動態(tài)攔截控制器方法的執(zhí)行。本文將詳細講講SpringMVC中攔截器的概念及入門案例，感興趣的可以嘗試一下
    2023-01-01

最新評論