springboot集成rabbitmq

🕗 发布于 2024-09-22 19:52 java-rabbitmq spring boot rabbitmq

开发环境

jdk:1.8
spring-rabbit:2.3.9
springboot:2.4.8

项目目录

配置文件

spring:
  datasource:
    type: com.zaxxer.hikari.HikariDataSource
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/library?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8&rewriteBatchedStatements=true
    username: root
    password: 123456
  rabbitmq:
    host: localhost
    port: 5675
    publisher-confirm-type: correlated # 开启确认模式
    publisher-returns: true # 开启退回模式
    listener:
      simple:
        acknowledge-mode: manual # 手动确认
        concurrency: 10 # 并发数
        max-concurrency: 20 # 最大并发数


  jpa:
    hibernate:
      ddl-auto: update
    properties:
      hibernate:
        dialect: org.hibernate.dialect.MySQL8Dialect

server:
  port: 8112

第一种模式

/**
 * @Description 1. "Hello World!"
 * @Author WuYiLong
 * @Date 2024/8/14 15:34
 */
@Slf4j
@Service
public class FirstModeExample {

    public static final String QUEUE = "firstModeQueue";
    public static final String EXCHANGE = "firstModeExchange";
    public static final String ROUTE_KEY = "firstModeRouteKey";

    @RabbitListener(bindings = {@QueueBinding(value = @Queue(name = QUEUE), exchange = @Exchange(name = EXCHANGE), key = {ROUTE_KEY})})
    public void consumer(String data, Channel channel, Message message) throws IOException {
        log.info("第一种模式：生产者-队列-消费者，数据：{}",data);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
}

第二种模式

/**
 * @Description 2. Work Queues
 * @Author WuYiLong
 * @Date 2024/8/14 16:12
 */
@Slf4j
@Service
public class SecondModeExample {

    public static final String QUEUE = "secondModeQueue";
    public static final String EXCHANGE = "secondModeExchange";
    public static final String ROUTE_KEY = "secondModeRouteKey";

    @RabbitListener(bindings = @QueueBinding(value =@Queue(name = QUEUE),exchange = @Exchange(name = EXCHANGE),key = ROUTE_KEY))
    public void consumer1(String data, Channel channel, Message message) throws IOException {
        log.info("消费者1正在消费:{}",data);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

    @RabbitListener(bindings = @QueueBinding(value =@Queue(name = QUEUE),exchange = @Exchange(name = EXCHANGE),key = ROUTE_KEY))
    public void consumer2(String data, Channel channel, Message message) throws IOException {
        log.info("消费者2正在消费:{}",data);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

}

第三种模式

/**
 * @Description 3. Publish/Subscribe
 * @Author WuYiLong
 * @Date 2024/8/14 16:37
 */
@Slf4j
@Service
public class ThirdModeExample {

    public static final String QUEUE1 = "thirdModeQueue1";
    public static final String QUEUE2 = "thirdModeQueue2";
    public static final String EXCHANGE = "thirdModeExchange";

    @RabbitListener(bindings = @QueueBinding(value = @Queue(QUEUE1), exchange = @Exchange(name = EXCHANGE,type = ExchangeTypes.FANOUT)))
    public void consumer1(String data, Channel channel, Message message) throws Exception {
        log.info("consumer1: {}", data);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(bindings = @QueueBinding(value = @Queue(QUEUE2), exchange = @Exchange(name=EXCHANGE,type = ExchangeTypes.FANOUT)))
    public void consumer2(String data, Channel channel, Message message) throws Exception {
        log.info("consumer2: {}", data);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

}

第四种模式

/**
 * @Description 4. Routing
 * @Author WuYiLong
 * @Date 2024/8/15 14:05
 */
@Slf4j
@Service
public class FourModeExample {

    public static final String QUEUE1 = "fourModeQueue1";
    public static final String QUEUE2 = "fourModeQueue2";
    public static final String EXCHANGE = "fourModeExchange";
    public static final String ROUTING_KEY1 = "fourModeRoutingKey1";
    public static final String ROUTING_KEY2 = "fourModeRoutingKey2";
    public static final String ROUTING_KEY3 = "fourModeRoutingKey3";

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = QUEUE1),exchange = @Exchange(value = EXCHANGE),key = {ROUTING_KEY1,ROUTING_KEY2,ROUTING_KEY3}))
    public void consumer1(String data, Channel channel, Message message) throws Exception {
        log.info("consumer1->{}", data);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = QUEUE2),exchange = @Exchange(value = EXCHANGE),key = {ROUTING_KEY1}))
    public void consumer2(String data, Channel channel, Message message) throws Exception {
        log.info("consumer2->{}", data);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

}

第五种模式

/**
 * @Description 5. Topics
 * @Author WuYiLong
 * @Date 2024/8/16 10:39
 */
@Slf4j
@Service
public class FiveModeExample {

    public static final String QUEUE1 = "queue1";
    public static final String QUEUE2 = "queue2";
    public static final String EXCHANGE = "exchange";
    public static final String ROUTING_KEY1 = "*.a.*";
    public static final String ROUTING_KEY2 = "*.*.b";
    public static final String ROUTING_KEY3 = "c.#";

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = QUEUE1),exchange = @Exchange(name = EXCHANGE,type = ExchangeTypes.TOPIC),key = ROUTING_KEY1))
    public void consumer1(String data, Channel channel, Message message) throws Exception {
        log.info("consumer1: {}", data);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = QUEUE2),exchange = @Exchange(name = EXCHANGE,type = ExchangeTypes.TOPIC),key = {ROUTING_KEY2,ROUTING_KEY3}))
    public void consumer2(String data, Channel channel, Message message) throws Exception {
        log.info("consumer2: {}", data);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }


}

第六种模式

注意：客户端和服务器端不能在同一个服务，必须分开两个服务写

要不会造成不断循环

服务器端

配置类

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory);
rabbitTemplate.setConfirmCallback(new DefaultConfirmCallback());
rabbitTemplate.setReturnsCallback(new DefaultReturnsCallback());
rabbitTemplate.setMandatory(true); //设置为true，才会触发DefaultReturnsCallback方法
//        rabbitTemplate.setChannelTransacted(true); // 支持事务,但使用RPC模式时需要关闭
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());  // 消息转换器，支持转换对象
return rabbitTemplate;
}

队列和交换机的绑定

/**
 * @Description
 * @Author WuYiLong
 * @Date 2024/8/19 17:24
 */
@Configuration
public class RabbitRpcConfig {

    public static final String REPLY_QUEUE = "sixModeReplyQueue";
    public static final String RPC_QUEUE = "sixModeRpcQueue";
    public static final String EXCHANGE = "sixModeExchange";
    public static final String RPC_ROUTING_KEY = "sixModeRpcRoutingKey";
    public static final String REPLY_ROUTING_KEY = "sixModeReplyRoutingKey";

    @Bean
    public Queue replyQueue() {
        return QueueBuilder.durable(REPLY_QUEUE).build();
    }

    @Bean
    public Exchange rpcExchange() {
        return ExchangeBuilder.topicExchange(EXCHANGE).build();
    }

    @Bean
    public Binding replyBinding() {
        return BindingBuilder.bind(replyQueue()).to(rpcExchange()).with(REPLY_ROUTING_KEY).noargs();
    }

    @Bean
    public Queue rpcQueue() {
        return QueueBuilder.durable(RPC_QUEUE).build();
    }

    @Bean
    public Binding rpcBinding() {
        return BindingBuilder.bind(rpcQueue()).to(rpcExchange()).with(RPC_ROUTING_KEY).noargs();
    }


}

监听类

用于监听客户端发送过来的消息

/**
 * @Description rpc模式服务器端
 * @Author WuYiLong
 * @Date 2024/8/16 14:31
 */
@Slf4j
@Service
public class SixModeExample {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RabbitListener(queues = RabbitRpcConfig.RPC_QUEUE)
    public void consumer(Channel channel,Message message) throws IOException {
        log.info("consumer->{}", StringUtils.toEncodedString(message.getBody(), StandardCharsets.UTF_8));
        Message build = MessageBuilder.withBody("已收到消息了".getBytes(StandardCharsets.UTF_8)).build();
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        String correlationId = message.getMessageProperties().getCorrelationId();
        CorrelationData correlationData = new CorrelationData(correlationId);
        rabbitTemplate.sendAndReceive(RabbitRpcConfig.EXCHANGE, RabbitRpcConfig.REPLY_ROUTING_KEY,build ,correlationData);

    }
}

客户端

配置类

/**
 * @Description
 * @Author WuYiLong
 * @Date 2024/8/19 17:20
 */
@Configuration
public class RabbitClientConfig {

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer();
        simpleMessageListenerContainer.setConnectionFactory(connectionFactory);
        simpleMessageListenerContainer.setQueueNames(RabbitRpcConfig.REPLY_QUEUE); // 监听的队列
        simpleMessageListenerContainer.setMessageListener(rabbitTemplate(connectionFactory));
        return simpleMessageListenerContainer;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());  // 消息转换器，支持转换对象
        rabbitTemplate.setReplyAddress(RabbitRpcConfig.REPLY_QUEUE); // 使用 RabbitTemplate发送和接收消息,并设置回调队列地址
        rabbitTemplate.setReplyTimeout(60000);
        return rabbitTemplate;
    }
}

客户端发送消息

/**
 * @Description
 * @Author WuYiLong
 * @Date 2024/8/19 17:30
 */
@Api(tags = "rpc-客户端api")
@RestController
@RequestMapping(value = "rpcClient")
public class RpcClientController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @ApiOperation(value = "客户端发送消息")
    @GetMapping(value = "sendMessage")
    public void sendMessage(String msg) {
        Message message = MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8)).build();
        Message result = rabbitTemplate.sendAndReceive(RabbitRpcConfig.EXCHANGE, RabbitRpcConfig.RPC_ROUTING_KEY, message);
        if(result != null) {
            String correlationId = message.getMessageProperties().getCorrelationId();
            String returnCorrelationId = result.getMessageProperties().getHeader("spring_returned_message_correlation");
            if(correlationId.equals(returnCorrelationId)) {
                System.out.println("已确认该消息由该客户端发送");
            }

            System.out.println(new String(result.getBody()));

        }
    }
}

死信队列(也可以当延迟队列使用)

队列和交换机的设置

  /********************************测试死信队列****************************************/

    @Bean
    public Queue ordinaryQueue() {
        return QueueBuilder.durable(DeadLetterQueueExample.ORDINARY_QUEUE).ttl(5000).deadLetterExchange(DeadLetterQueueExample.DEAD_LETTER_EXCHANGE).deadLetterRoutingKey(DeadLetterQueueExample.DEAD_LETTER_ROUTING_KEY).build();
    }

    @Bean
    public Exchange ordinaryExchange() {
        return ExchangeBuilder.directExchange(DeadLetterQueueExample.ORDINARY_EXCHANGE).durable(true).build();
    }

    @Bean
    public Binding ordinaryBinding() {
        return BindingBuilder.bind(ordinaryQueue()).to(ordinaryExchange()).with(DeadLetterQueueExample.ORDINARY_ROUTING_KEY).noargs();
    }

    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable(DeadLetterQueueExample.DEAD_LETTER_QUEUE).build();
    }

    @Bean
    public Exchange deadLetterExchange() {
        return ExchangeBuilder.directExchange(DeadLetterQueueExample.DEAD_LETTER_EXCHANGE).durable(true).build();
    }

    @Bean
    public Binding deadLetterBinding() {
        return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(DeadLetterQueueExample.DEAD_LETTER_ROUTING_KEY).noargs();
    }

监听类

只需要监听deadLetterQueue队列的消息

/**
 * @Description 死信队列栗子
 * @Author WuYiLong
 * @Date 2024/8/20 9:41
 */
@Slf4j
@Service
public class DeadLetterQueueExample {

    public static final String ORDINARY_QUEUE = "ordinaryQueue";
    public static final String ORDINARY_EXCHANGE = "ordinaryExchange";
    public static final String ORDINARY_ROUTING_KEY = "ordinaryRoutingKey";
    public static final String DEAD_LETTER_QUEUE = "deadLetterQueue";
    public static final String DEAD_LETTER_EXCHANGE = "deadLetterExchange";
    public static final String DEAD_LETTER_ROUTING_KEY = "deadLetterRoutingKey";

    @RabbitListener(queues = DEAD_LETTER_QUEUE)
    public void deadConsumer(Channel channel, Message message) throws IOException {
       log.info("deadConsumer message: {}", new String(message.getBody()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

}

测试类

/**
 * @Description
 * @Author WuYiLong
 * @Date 2024/8/20 10:46
 */
@RunWith(SpringRunner.class)
@SpringBootTest
public class DeadLetterQueueExampleTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testDeadLetterQueueExample() throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            String msg = "java是一门计算机语言"+i;
//            rabbitTemplate.convertAndSend(DeadLetterQueueExample.ORDINARY_EXCHANGE,DeadLetterQueueExample.ORDINARY_ROUTING_KEY,msg);

            Message message = new Message(msg.getBytes(StandardCharsets.UTF_8));
//            message.getMessageProperties().setExpiration("5000"); // 如果消息设置了过期时间，队列就不用设置了.ttl(5000)
            rabbitTemplate.send(DeadLetterQueueExample.ORDINARY_EXCHANGE,DeadLetterQueueExample.ORDINARY_ROUTING_KEY,message);
        }
        TimeUnit.SECONDS.sleep(20);

    }
}

仓库地址

github

原文地址：https://blog.csdn.net/weixin_44830665/article/details/142414820

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：pytorch 同步机制
下一篇：js进阶——函数作用域和块作用域

栈和队列（选择题）
栈的存储结构可以分为和。顺序存储的栈称为，链式存储的栈称为。（1）顺序栈顺序栈：利用一组地址连续的存储单元存放自栈底到栈顶的数据元素，同时附设一个指针（top）指示当前栈顶位置。栈空条件：顺序栈的数组
阅读更多2024-09-22
C++从入门到起飞之——多态全方位剖析！
多态是⼀个继承关系的下的类对象，去调⽤同⼀函数，产⽣了不同的⾏为。⽐如Student继承了 Person。Person对象买票全价，Student对象优惠买票。
阅读更多2024-09-22
9月16日笔记
当安全主体访问该安全对象时，Windows会检查安全主体的 SID和安全对象 DACL中的 ACE 配置策略，根据找到的 ACE 配置策略对安全主体的访问行为允许或拒绝。MA，允许的最大值;选中预先创
阅读更多2024-09-22
synchronized锁的升级过程
synchronized锁的升级过程
阅读更多2024-09-22
电脑远程启动Wol
Wake On lan
阅读更多2024-09-22
专业学习｜《随机过程》学习笔记（二）（定义、分类及相关过程）
本片博客继上一篇学习指引文之后，强调介绍了随机过程的分类，让读者能对随机过程、随机过程的分类有一个系统了解，其中还补充了一些基础知识以便读者了解推导过程。
阅读更多2024-09-22
汇川AM600 与 codesys opc server 通讯并建立 OPCserver 的方法
AM600 OPC Server 架构。
阅读更多2024-09-22
VM-Ubantu中使用vscode头文件报错——解决办法
VM-Ubantu中使用vscode头文件报错——解决办法。使用快捷键shift+ctrl+p，点击Edit Config(JSON)，并将下面的代码复制进去即可
阅读更多2024-09-22
Hive企业级调优[2]—— 测试用表
文件上传到 hadoop12 节点的。文件上传到 hadoop12 节点的。文件上传到 hadoop12 节点的。文件上传到 hadoop12 节点的。注：需要下载源数据的，评论区私俺。目录，并执行以
阅读更多2024-09-22
力扣无重复字符的最长子串
使用滑动窗口结合双指针，滑动窗口的表示采用哈希表存储字母数量信息的方式在窗口滑动的过程中维护最大不重复子串长度。
阅读更多2024-09-22