脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → springboot集成RabbitMq

springboot2.5.6集成RabbitMq实现Topic主题模式(推荐)

 更新时间：2021年11月04日 09:36:58   作者：Scarlet-Max  
这篇文章主要介绍了springboot2.5.6集成RabbitMq实现Topic主题模式(推荐)，pom.xml引入依赖和常量类创建，本文通过实例代码给大家介绍的非常详细，需要的朋友参考下吧

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 # 下次重试间隔的倍数，此处是2即下次重试间隔是上次的2倍
      exchange: topic.exchange # 缺省的交换机名称，此处配置后，发送消息如果不指定交换机就会使用这个
    publisher-confirm-type: correlated # 生产者确认机制，确保消息会正确发送，如果发送失败会有错误回执，从而触发重试
    publisher-returns: true
    listener:
      type: simple
      simple:
        acknowledge-mode: manual
        prefetch: 1 # 限制每次发送一条数据。
        concurrency: 3 # 同一个队列启动几个消费者
        max-concurrency: 3 # 启动消费者最大数量
        # 重试策略相关配置
        retry:
          enabled: true # 是否支持重试
          max-attempts: 5
          stateless: false
          multiplier: 1.0 # 时间策略乘数因子
          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.常量类创建

/**
 * @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();
        //发布订阅模式
        return ExchangeBuilder.fanoutExchange(RabbitMqConstants.EXCHANGE_NAME).durable(true).build();
    }

    /**
     *  声明队列
     *  new Queue(QUEUE_EMAIL,true,false,false)
     *  durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列
     *  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();
    }

    /**
     * 如果需要在生产者需要消息发送后的回调，
     * 需要对rabbitTemplate设置ConfirmCallback对象，
     * 由于不同的生产者需要对应不同的ConfirmCallback，
     * 如果rabbitTemplate设置为单例bean，
     * 则所有的rabbitTemplate实际的ConfirmCallback为最后一次申明的ConfirmCallback。
     * @return
     */
    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate() {
         RabbitTemplate template = new RabbitTemplate(connectionFactory);
        return template;
    }
}

5.Rabbit工具类创建

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;
        //这是是设置回调能收到发送到响应
        rabbitTemplate.setConfirmCallback(this);
        //如果设置备份队列则不起作用
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(this);
    }

    /**
     * 回调确认
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if(ack){
            log.info("消息发送成功：correlationData({}),ack({}),cause({})",correlationData,ack,cause);
        }else{
            log.info("消息发送失败：correlationData({}),ack({}),cause({})",correlationData,ack,cause);
        }
    }

    /**
     * 消息发送到转换器的时候没有对列,配置了备份对列该回调则不生效
     * @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);
    }

    /**
     * 发送到指定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、消息内容
     */
    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创建

public interface TestService {

    String sendTest1(String content);

    String sendTest2(String content);
}

7.impl实现

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+"***************发送成功*****************");
        return "发送成功！";
    }

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

8.监听类

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 {
            //手动确认消息已经被消费
            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 {
            //手动确认消息已经被消费
            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());
    }
    /**
     * 发送消息test2
     * @param content
     * @return
     */
    @PostMapping(value = "/finance2")
    public String sendTest2(@RequestBody String content) {
        return testService.sendTest2(content);
    }

}

在这里插入图片描述

到此这篇关于springboot2.5.6集成RabbitMq实现Topic主题模式的文章就介绍到这了,更多相关springboot集成RabbitMq内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • Java中CopyOnWriteArrayList的使用解析

    Java中CopyOnWriteArrayList的使用解析

    这篇文章主要介绍了Java中CopyOnWriteArrayList的使用解析,CopyOnWriteArrayList适合使用在读操作远远大于写操作的场景里,比如缓存,它不存在扩容的概念,每次写操作都要复制一个副本,在副本的基础上修改后改变Array引用,需要的朋友可以参考下
    2023-12-12
  • JAVA8 Stream流中的reduce()方法详解

    JAVA8 Stream流中的reduce()方法详解

    reduce 操作可以实现从Stream中生成一个值，其生成的值不是随意的，而是根据指定的计算模型，这篇文章主要介绍了JAVA8 Stream流中的reduce()方法详解,需要的朋友可以参考下
    2023-02-02
  • java项目jar包与jdk的版本不兼容的问题解决

    java项目jar包与jdk的版本不兼容的问题解决

    这篇文章主要介绍了java项目jar包与jdk的版本不兼容的问题解决,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2019-10-10
  • Java System类详解_动力节点Java学院整理

    Java System类详解_动力节点Java学院整理

    System类是jdk提供的一个工具类，有final修饰，不可继承，由名字可以看出来，其中的操作多数和系统相关。这篇文章主要介绍了Java System类详解_动力节点Java学院整理,需要的朋友可以参考下
    2017-04-04
  • 微信支付java版本之JSAPI支付+发送模板消息

    微信支付java版本之JSAPI支付+发送模板消息

    这篇文章主要介绍了微信支付java版本之JSAPI支付，发送模板消息，感兴趣的小伙伴们可以参考一下
    2016-08-08
  • 浅谈MyBatis循环Map(高级用法)

    浅谈MyBatis循环Map(高级用法)

    这篇文章主要介绍了浅谈MyBatis循环Map(高级用法)，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2020-09-09
  • java 使用HttpURLConnection发送数据简单实例

    java 使用HttpURLConnection发送数据简单实例

    这篇文章主要介绍了java 使用HttpURLConnection发送数据简单实例的相关资料,需要的朋友可以参考下
    2017-06-06
  • Java数据结构之位图的简单实现和使用

    Java数据结构之位图的简单实现和使用

    位图, 是一种非常常见的结构, 它使用每个二进制位来存放一个值的状态, 就类似于 Java 当中 HashSet 存储元素的功能。本文主要来介绍一下位图的简单实现和使用，需要的可以参考一下
    2023-05-05
  • Java开启JMX远程监控服务配置

    Java开启JMX远程监控服务配置

    这篇文章主要为大家介绍了Java开启JMX远程监控的服务配置，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
    2023-05-05
  • 解读CommandLineRunner和@PostConstruct区别与应用场景

    解读CommandLineRunner和@PostConstruct区别与应用场景

    这篇文章主要介绍了解读CommandLineRunner和@PostConstruct区别与应用场景,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2023-12-12

最新评论