spring整合RabbitMQ

今天就来康康spring怎么整合RabbitMQ

　　注意一点，在发送消息的时候对template进行配置mandatory=true保证监听有效

　 生产端还可以配置其他属性，比如发送重试，超时时间、次数、间隔等

消费端核心配置

  1. 首先配置手工确认模式，用于ACK的手工处理，这样我们可以保证消息的可靠性送达，或者在消费端消费失败的时候可以做到重回队列、根据业务记录日志等处理
  2. 可以设置消费端的监听个数和最大个数，用于控制消费端的并发情况

@RabbitListener注解的使用

消费端监听@RabbitListener注解，这个对于在实际工作中非常的好用

@RabbitListener是一个组合注解，里面可以注

解配置（@QueueBinding、@Queue、@Exchange）直接通过这个组合注解一次性搞定消费端交换机、队列、绑定、路由、并且配置监听功能等

注：由于类配置写在代码里非常不友好，所以强烈建议大家使用配置文件配置

创建一个

rabbitmq-common通用模板项目

放公用实体类

package com.javaxh.rabbitmqcommon.entity;
import java.io.Serializable;
public class Order implements Serializable {

    private String id;
    private String name;

    public Order() {
    }
    public Order(String id, String name) {
        super();
        this.id = id;
        this.name = name;
    }
    public String getId() {
        return id;
    }
    public void setId(String id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
}

创建

rabbitmq-springcloud-consumer子项目

 pom文件中添加依赖：

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

        <dependency>
            <groupId>com.javaxh</groupId>
            <artifactId>rabbitmq-common</artifactId>
            <version>0.0.1-SNAPSHOT</version>
        </dependency>

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

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

application.properties配置

spring.rabbitmq.addresses=192.168.239.131:5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=15000

server.port=80
server.servlet.context-path=/

spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10

spring.rabbitmq.listener.order.queue.name=queue-2
spring.rabbitmq.listener.order.queue.durable=true
spring.rabbitmq.listener.order.exchange.name=exchange-2
spring.rabbitmq.listener.order.exchange.durable=true
spring.rabbitmq.listener.order.exchange.type=topic
spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
spring.rabbitmq.listener.order.key=springboot.*

RabbitReceiver

package com.javaxh.rabbitmqspringcloudconsumer.conusmer;
import com.javaxh.rabbitmqcommon.entity.Order;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.Map;
@Component
public class RabbitReceiver {
   @RabbitListener(bindings = @QueueBinding(
         value = @Queue(value = "queue-1",
         durable="true"),
         exchange = @Exchange(value = "exchange-1",
         durable="true",
         type= "topic",
         ignoreDeclarationExceptions = "true"),
         key = "springboot.*"
         )
   )
   @RabbitHandler
   public void onMessage(Message message, Channel channel) throws Exception {
      System.err.println("--------------------------------------");
      System.err.println("消费端Payload: " + message.getPayload());
      Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
      //手工ACK
      channel.basicAck(deliveryTag, false);
   }

   /**
    *
    *     spring.rabbitmq.listener.order.queue.name=queue-2
      spring.rabbitmq.listener.order.queue.durable=true
      spring.rabbitmq.listener.order.exchange.name=exchange-1
      spring.rabbitmq.listener.order.exchange.durable=true
      spring.rabbitmq.listener.order.exchange.type=topic
      spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions=true
      spring.rabbitmq.listener.order.key=springboot.*
    * @param order
    * @param channel
    * @param headers
    * @throws Exception
    */
   @RabbitListener(bindings = @QueueBinding(
         value = @Queue(value = "${spring.rabbitmq.listener.order.queue.name}",
         durable="${spring.rabbitmq.listener.order.queue.durable}"),
         exchange = @Exchange(value = "${spring.rabbitmq.listener.order.exchange.name}",
         durable="${spring.rabbitmq.listener.order.exchange.durable}",
         type= "${spring.rabbitmq.listener.order.exchange.type}",
         ignoreDeclarationExceptions = "${spring.rabbitmq.listener.order.exchange.ignoreDeclarationExceptions}"),
         key = "${spring.rabbitmq.listener.order.key}"
         )
   )
   @RabbitHandler
   public void onOrderMessage(@Payload Order order,
         Channel channel,
         @Headers Map<String, Object> headers) throws Exception {
      System.err.println("--------------------------------------");
      System.err.println("消费端order: " + order.getId());
      Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
      //手工ACK
      channel.basicAck(deliveryTag, false);
   }

}

MainConfig

package com.javaxh.rabbitmqspringcloudconsumer;

import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

@Configuration
@ComponentScan({"com.javaxh.rabbitmqspringcloudconsumer.*"})
public class MainConfig {

}

创建

rabbitmq-springcloud-producer子项目

在pom中添加依赖

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

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

        <dependency>
            <groupId>com.javaxh</groupId>
            <artifactId>rabbitmq-common</artifactId>
            <version>0.0.1-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

application.properties配置

spring.rabbitmq.addresses=192.168.239.131:5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=15000

spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true
RabbitSender
package com.javaxh.rabbitmqspringcloudproducer.producer;

import com.javaxh.rabbitmqcommon.entity.Order;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class RabbitSender {

   //自动注入RabbitTemplate模板类
   @Autowired
   private RabbitTemplate rabbitTemplate;

   //回调函数: confirm确认
   final ConfirmCallback confirmCallback = new ConfirmCallback() {
      @Override
      public void confirm(CorrelationData correlationData, boolean ack, String cause) {
         System.err.println("correlationData: " + correlationData);
         System.err.println("ack: " + ack);
         if(!ack){
            System.err.println("异常处理....");
         }
      }
   };

   //回调函数: return返回
   final ReturnCallback returnCallback = new ReturnCallback() {
      @Override
      public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText,
            String exchange, String routingKey) {
         System.err.println("return exchange: " + exchange + ", routingKey: "
            + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);
      }
   };

   //发送消息方法调用: 构建Message消息
   public void send(Object message, Map<String, Object> properties) throws Exception {
      MessageHeaders mhs = new MessageHeaders(properties);
      Message msg = MessageBuilder.createMessage(message, mhs);
      rabbitTemplate.setConfirmCallback(confirmCallback);
      rabbitTemplate.setReturnCallback(returnCallback);
      //id + 时间戳 全局唯一
      CorrelationData correlationData = new CorrelationData("1234567890");
      rabbitTemplate.convertAndSend("exchange-1", "springboot.abc", msg, correlationData);
   }

   //发送消息方法调用: 构建自定义对象消息
   public void sendOrder(Order order) throws Exception {
      rabbitTemplate.setConfirmCallback(confirmCallback);
      rabbitTemplate.setReturnCallback(returnCallback);
      //id + 时间戳 全局唯一
      CorrelationData correlationData = new CorrelationData("0987654321");
      rabbitTemplate.convertAndSend("exchange-2", "springboot.def", order, correlationData);
   }

}
MainConfig 
package com.javaxh.rabbitmqspringcloudproducer;

import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

@Configuration
@ComponentScan({"com.javaxh.rabbitmqspringcloudproducer.*"})
public class MainConfig {

}

测试代码：

RabbitmqSpringcloudProducerApplicationTests
package com.javaxh.rabbitmqspringcloudproducer;
import com.javaxh.rabbitmqcommon.entity.Order;
import com.javaxh.rabbitmqspringcloudproducer.producer.RabbitSender;
import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
class RabbitmqSpringcloudProducerApplicationTests {

    @Autowired
    private RabbitSender rabbitSender;

    private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    @Test
    public void testSender1() throws Exception {
        Map<String, Object> properties = new HashMap<>();
        properties.put("number", "12345");
        properties.put("send_time", simpleDateFormat.format(new Date()));
        rabbitSender.send("Hello RabbitMQ For Spring Boot!", properties);
    }

    @Test
    public void testSender2() throws Exception {
        Order order = new Order("001", "第一个订单");
        rabbitSender.sendOrder(order);
    }

}

运行效果：

启动consumer项目，然后在运行测试类中的方法

运行公用模板类传送一个实体类给方法：

谢谢观看！！！

原文地址：https://www.cnblogs.com/huangting/p/12013306.html

时间： 2024-10-03 14:43:00

spring整合RabbitMQ的相关文章

spring 整合rabbitMQ

<!-- 配置邮件消息队列监听 --> <bean id="maillistener" class="cn.xdf.wlyy.listener.Maillistener" name="maillistener"> <property name="mailManager" ref="mailManager"></property> <property na

022 spring与Rabbitmq整合

一 .概述 本次我们使用spring帮助我们完成Rabbitmq的使用. 二 .环境的搭建 本次使用springboot的jar文件帮助整合rabbitmq,但是本质上还是使用spring的方式进行整合. <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.

Spring Boot （十三）： Spring Boot 整合 RabbitMQ

1. 前言 RabbitMQ 是一个消息队列,说到消息队列,大家可能多多少少有听过,它主要的功能是用来实现应用服务的异步与解耦,同时也能起到削峰填谷.消息分发的作用. 消息队列在比较主要的一个作用是用来做应用服务的解耦,消息从消息的生产者传递到消息队列,消费者从消息队列中获取消息并进行消费,生产者不需要管是谁在消费消息,消费者也无需关注消息是由谁来生产的.在分布式的系统中,消息队列也会被用在其他地方,比如分布式事务的支持,代表如阿里开源的 RocketMQ . 当然,我们本篇文章的主角还是 Ra

Spring Boot (5) 整合 RabbitMQ

一.前言 RabbitMQ是实现了AMQP(高级消息队列协议)的开源消息中间件,RabbitMQ服务器是用Erlang(面向并发的编程语言)编写的. RabbitMQ官网下载地址:https://www.rabbitmq.com/download.html Docker部署则执行如下命令即可 # RABBITMQ_DEFAULT_USER:用户名,RABBITMQ_DEFAULT_PASS:密码 这里修改为自己的即可 docker run -d --name rabbitmq -p 5672:5

Rabbitmq与spring整合之重要组件介绍——AMQP声明式配置&amp;RabbitTemplate组件

上一节是使用rabbitAdmin的管理组件进行声明队列,交换器,绑定等操作,本节则是采用AMQP声明式配置来声明这些东西.AMQP声明主要是通过@Bean注解进行的. 配置: 1 package com.zxy.demo.config; 2 3 import org.springframework.amqp.core.Binding; 4 import org.springframework.amqp.core.BindingBuilder; 5 import org.springframew

springboot学习笔记-6 springboot整合RabbitMQ

一 RabbitMQ的介绍 RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apache公司的),RocketMQ(阿里巴巴公司的,现已经转让给apache). 消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下: 从上图可看出,对于消息队列来说,生产者,消息队列,消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息

springboot系列-springboot整合RabbitMQ

一 RabbitMQ的介绍 RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apache公司的),RocketMQ(阿里巴巴公司的,现已经转让给apache). 消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下: 从上图可看出,对于消息队列来说,生产者,消息队列,消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息

SpringBoot整合RabbitMQ之整合配置篇

实战背景:RabbitMQ实战第一阶段-RabbitMQ的官网拜读已经结束了,相信诸位童鞋或多或少都能入了个门,如果还是觉得迷迷糊糊似懂非懂的,那我建议诸位可以亲自去拜读拜读官网的技术手册或者看多几篇我的视频跟源码!因为接下来我们将进入第二阶段,即应用实战阶段(备注:第一阶段的内容主要以视频的形式分享,感兴趣的童鞋可以加QQ群:583522159 自行获取) 实战分析:应用实战,当然是指真正的在企业级项目中的应用.在这一阶段中,我将以目前流行的微服务架构为奠基,整合RabbitMQ实现项目中常见

SpringBoot整合RabbitMQ之发送接收消息实战

实战前言 前几篇文章中,我们介绍了SpringBoot整合RabbitMQ的配置以及实战了Spring的事件驱动模型,这两篇文章对于我们后续实战RabbitMQ其他知识要点将起到奠基的作用的.特别是Spring的事件驱动模型,当我们全篇实战完毕RabbitMQ并大概了解一下RabbitMQ相关组件的源码时,会发现其中的ApplicationEvent.ApplicationListener.ApplicationEventPublisher跟RabbitMQ的Message.Listener.R