springboot2.5.6集成RabbitMq實現(xiàn)Topic主題模式(推薦)

更新時間：2021年11月04日 09:36:58 作者：Scarlet-Max

這篇文章主要介紹了springboot2.5.6集成RabbitMq實現(xiàn)Topic主題模式(推薦)，pom.xml引入依賴和常量類創(chuàng)建，本文通過實例代碼給大家介紹的非常詳細，需要的朋友參考下吧

1.application.yml

server:
  port: 8184
spring:
  application:
    name: rabbitmq-demo
  rabbitmq:
    host: 127.0.0.1 # ip地址
    port: 5672
    username: admin # 連接賬號
    password: 123456 # 連接密碼
    template:
      retry:
        enabled: true # 開啟失敗重試
        initial-interval: 10000ms # 第一次重試的間隔時長
        max-interval: 300000ms # 最長重試間隔，超過這個間隔將不再重試
        multiplier: 2 # 下次重試間隔的倍數(shù)，此處是2即下次重試間隔是上次的2倍
      exchange: topic.exchange # 缺省的交換機名稱，此處配置后，發(fā)送消息如果不指定交換機就會使用這個
    publisher-confirm-type: correlated # 生產(chǎn)者確認機制，確保消息會正確發(fā)送，如果發(fā)送失敗會有錯誤回執(zhí)，從而觸發(fā)重試
    publisher-returns: true
    listener:
      type: simple
      simple:
        acknowledge-mode: manual
        prefetch: 1 # 限制每次發(fā)送一條數(shù)據(jù)。
        concurrency: 3 # 同一個隊列啟動幾個消費者
        max-concurrency: 3 # 啟動消費者最大數(shù)量
        # 重試策略相關配置
        retry:
          enabled: true # 是否支持重試
          max-attempts: 5
          stateless: false
          multiplier: 1.0 # 時間策略乘數(shù)因子
          initial-interval: 1000ms
          max-interval: 10000ms
        default-requeue-rejected: true

2.pom.xml引入依賴

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

3.常量類創(chuàng)建

/**
 * @author kkp
 * @ClassName RabbitMqConstants
 * @date 2021/11/3 14:16
 * @Description
 */
public class RabbitMqConstants {
    public final static String TEST1_QUEUE = "test1-queue";

    public final static String TEST2_QUEUE = "test2-queue";

    public final static String EXCHANGE_NAME = "test.topic.exchange";
    /**
     * routingKey1
     */
    public final static String TOPIC_TEST1_ROUTINGKEY = "topic.test1.*";

    public final static String TOPIC_TEST1_ROUTINGKEY_TEST = "topic.test1.test";
    /**
     * routingKey1
     */
    public final static String TOPIC_TEST2_ROUTINGKEY = "topic.test2.*";

    public final static String TOPIC_TEST2_ROUTINGKEY_TEST = "topic.test2.test";
}

4.配置Configuration

import com.example.demo.common.RabbitMqConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

/**
 * @author kkp
 * @ClassName RabbitMqConfig
 * @date 2021/11/3 14:16
 * @Description
 */
@Slf4j
@Configuration
public class RabbitMqConfig {
    @Autowired
    private CachingConnectionFactory connectionFactory;

    /**
     *  聲明交換機
     */
    @Bean(RabbitMqConstants.EXCHANGE_NAME)
    public Exchange exchange(){
 //durable(true) 持久化，mq重啟之后交換機還在
        // Topic模式
        //return ExchangeBuilder.topicExchange(RabbitMqConstants.EXCHANGE_NAME).durable(true).build();
        //發(fā)布訂閱模式
        return ExchangeBuilder.fanoutExchange(RabbitMqConstants.EXCHANGE_NAME).durable(true).build();
    }

    /**
     *  聲明隊列
     *  new Queue(QUEUE_EMAIL,true,false,false)
     *  durable="true" 持久化 rabbitmq重啟的時候不需要創(chuàng)建新的隊列
     *  auto-delete 表示消息隊列沒有在使用時將被自動刪除 默認是false
     *  exclusive  表示該消息隊列是否只在當前connection生效,默認是false
     */
    @Bean(RabbitMqConstants.TEST1_QUEUE)
    public Queue esQueue() {
        return new Queue(RabbitMqConstants.TEST1_QUEUE);
    }

    /**
     *  聲明隊列
     */
    @Bean(RabbitMqConstants.TEST2_QUEUE)
    public Queue gitalkQueue() {
        return new Queue(RabbitMqConstants.TEST2_QUEUE);
    }

    /**
     *  TEST1_QUEUE隊列綁定交換機，指定routingKey
     */
    @Bean
    public Binding bindingEs(@Qualifier(RabbitMqConstants.TEST1_QUEUE) Queue queue,
                             @Qualifier(RabbitMqConstants.EXCHANGE_NAME) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY).noargs();
    }

    /**
     *  TEST2_QUEUE隊列綁定交換機，指定routingKey
     */
    @Bean
    public Binding bindingGitalk(@Qualifier(RabbitMqConstants.TEST2_QUEUE) Queue queue,
                                 @Qualifier(RabbitMqConstants.EXCHANGE_NAME) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY).noargs();
    }

    /**
     * 如果需要在生產(chǎn)者需要消息發(fā)送后的回調(diào)，
     * 需要對rabbitTemplate設置ConfirmCallback對象，
     * 由于不同的生產(chǎn)者需要對應不同的ConfirmCallback，
     * 如果rabbitTemplate設置為單例bean，
     * 則所有的rabbitTemplate實際的ConfirmCallback為最后一次申明的ConfirmCallback。
     * @return
     */
    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate() {
         RabbitTemplate template = new RabbitTemplate(connectionFactory);
        return template;
    }
}

5.Rabbit工具類創(chuàng)建

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
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 java.util.UUID;
/**
 * @author kkp
 * @ClassName RabbitMqUtils
 * @date 2021/11/3 14:21
 * @Description
 */
@Slf4j
@Component
public class RabbitMqUtils implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{

    private RabbitTemplate rabbitTemplate;

    /**
     * 構造方法注入
     */
    @Autowired
    public RabbitMqUtils(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        //這是是設置回調(diào)能收到發(fā)送到響應
        rabbitTemplate.setConfirmCallback(this);
        //如果設置備份隊列則不起作用
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(this);
    }

    /**
     * 回調(diào)確認
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if(ack){
            log.info("消息發(fā)送成功：correlationData({}),ack({}),cause({})",correlationData,ack,cause);
        }else{
            log.info("消息發(fā)送失?。篶orrelationData({}),ack({}),cause({})",correlationData,ack,cause);
        }
    }

    /**
     * 消息發(fā)送到轉換器的時候沒有對列,配置了備份對列該回調(diào)則不生效
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("消息丟失：exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
    }

    /**
     * 發(fā)送到指定Queue
     * @param queueName
     * @param obj
     */
    public void send(String queueName, Object obj){
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        this.rabbitTemplate.convertAndSend(queueName, obj, correlationId);
    }

    /**
     * 1、交換機名稱
     * 2、routingKey
     * 3、消息內(nèi)容
     */
    public void sendByRoutingKey(String exChange, String routingKey, Object obj){
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        this.rabbitTemplate.convertAndSend(exChange, routingKey, obj, correlationId);
    }
}

6.service創(chuàng)建

public interface TestService {

    String sendTest1(String content);

    String sendTest2(String content);
}

7.impl實現(xiàn)

import com.example.demo.common.RabbitMqConstants;
import com.example.demo.util.RabbitMqUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
 * @author kkp
 * @ClassName TestServiceImpl
 * @date 2021/11/3 14:24
 * @Description
 */
@Service
@Slf4j
public class TestServiceImpl implements TestService {

    @Autowired
    private RabbitMqUtils rabbitMqUtils;

    @Override
    public String sendTest1(String content) {
        rabbitMqUtils.sendByRoutingKey(RabbitMqConstants.EXCHANGE_NAME,
                RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY_TEST, content);
        log.info(RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY_TEST+"***************發(fā)送成功*****************");
        return "發(fā)送成功！";
    }

    @Override
    public String sendTest2(String content) {
        rabbitMqUtils.sendByRoutingKey(RabbitMqConstants.EXCHANGE_NAME,
                RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY_TEST, content);
        log.info(RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY_TEST+"***************發(fā)送成功*****************");
        return "發(fā)送成功！";
    }
}

8.監(jiān)聽類

import com.example.demo.common.RabbitMqConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
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.stereotype.Component;

import com.rabbitmq.client.Channel;

/**
 * @author kkp
 * @ClassName RabbitMqListener
 * @date 2021/11/3 14:22
 * @Description
 */

@Slf4j
@Component
public class RabbitMqListener {

    @RabbitListener(queues = RabbitMqConstants.TEST1_QUEUE)
    public void test1Consumer(Message message, Channel channel) {
        try {
            //手動確認消息已經(jīng)被消費
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            log.info("Counsoum1消費消息：" + message.toString() + "。成功！");
        } catch (Exception e) {
            e.printStackTrace();
            log.info("Counsoum1消費消息：" + message.toString() + "。失?。?);
        }
    }

    @RabbitListener(queues = RabbitMqConstants.TEST2_QUEUE)
    public void test2Consumer(Message message, Channel channel) {
        try {
            //手動確認消息已經(jīng)被消費
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            log.info("Counsoum2消費消息：" + message.toString() + "。成功！");
        } catch (Exception e) {
            e.printStackTrace();
            log.info("Counsoum2消費消息：" + message.toString() + "。失??！");
        }
    }

}

9.Controller測試

import com.example.demo.server.TestService;
import jdk.nashorn.internal.objects.annotations.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.Map;

/**
 * @author kkp
 * @ClassName TestController
 * @date 2021/11/3 14:25
 * @Description
 */
@Slf4j
@RestController
@RequestMapping("/enterprise")
public class TestController {

    @Autowired
    private TestService testService;

    @GetMapping("/finance")
    public String hello3(@RequestParam(required = false) Map<String, Object> params) {
        return testService.sendTest2(params.get("entId").toString());
    }
    /**
     * 發(fā)送消息test2
     * @param content
     * @return
     */
    @PostMapping(value = "/finance2")
    public String sendTest2(@RequestBody String content) {
        return testService.sendTest2(content);
    }

}

在這里插入圖片描述

到此這篇關于springboot2.5.6集成RabbitMq實現(xiàn)Topic主題模式的文章就介紹到這了,更多相關springboot集成RabbitMq內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

我勸你謹慎使用Spring中的@Scheduled注解
這篇文章主要介紹了Spring中的@Scheduled注解使用，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
2021-10-10
java ThreadPoolExecutor 并發(fā)調(diào)用實例詳解
這篇文章主要介紹了java ThreadPoolExecutor 并發(fā)調(diào)用實例詳解的相關資料,需要的朋友可以參考下
2017-05-05
Java 高并發(fā)四：無鎖詳細介紹
本文主要介紹Java 高并發(fā)無鎖的知識，這里整理了 1.無鎖類的原理詳解 2.無鎖類的使用的知識，并講解其原理，有需要的小伙伴可以參考下
2016-09-09
關于Spring中@Lazy注解的使用
這篇文章主要介紹了關于Spring中@Lazy注解的使用,@Lazy注解用于標識bean是否需要延遲加載,沒加注解之前主要容器啟動就會實例化bean,本文提供了部分實現(xiàn)代碼,需要的朋友可以參考下
2023-08-08
struts2過濾器和攔截器的區(qū)別分析
這篇文章主要介紹了struts2過濾器和攔截器的區(qū)別,簡單分析了struts2框架中過濾器和攔截器的概念與相關使用區(qū)別,需要的朋友可以參考下
2016-04-04
ArrayList集合初始化及擴容方式
這篇文章主要介紹了關于ArrayList集合初始化及擴容方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
2023-03-03
Spring @Profile注解實現(xiàn)多環(huán)境配置
這篇文章主要介紹了Spring @Profile注解實現(xiàn)多環(huán)境配置,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下
2020-04-04
分享JVM 的四種引用方式
這篇文章主要介紹了分享JVM 的四種引用方式，文章圍繞主題展開詳細的內(nèi)容介紹，具有一定的參考價值，需要的小伙伴可以參考一下
2022-07-07
java實現(xiàn)死鎖的示例代碼
本篇文章主要介紹了java實現(xiàn)死鎖的示例代碼，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧
2018-01-01
java圖片和文本同時提交到表單的實例代碼
在本篇文章里小編給大家整理的是關于java實現(xiàn)圖片和文本同時提交到表單的相關內(nèi)容，有需要的朋友們可以學習下。
2020-02-02