关键词:
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);
Spring Boot 集成测试。如何测试“删除”方法?
...【发布时间】:2021-12-1316:04:20【问题描述】:我正在使用SpringBoot2.5.6和JUnit4.13.2。我的任务是测试DELETE方法我的REST控制器:@RestControllerpublicclassDomainEndpointp 查看详情
springboot集成rabbitmq,eclipse开发集成rabbitmq,idea集成rabbitmq报错socketclose
java.net.SocketException:socketclosed atjava.net.SocketInputStream.socketRead0(NativeMethod) atjava.net.SocketInputStream.socketRead(SocketInputStream.java:116) 查看详情
rabbitmq第五篇:spring集成rabbitmq
前面几篇讲解了如何使用rabbitMq,这一篇主要讲解spring集成rabbitmq。 首先引入配置文件org.springframework.amqp,如下<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring 查看详情
rabbitmq第四篇:spring集成rabbitmq
前面几篇讲解了如何使用rabbitMq,这一篇主要讲解spring集成rabbitmq。 首先引入配置文件org.springframework.amqp,如下<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring 查看详情
springboot系列——7步集成rabbitmq
RabbitMQ是一种我们经常使用的消息中间件,通过RabbitMQ可以帮助我们实现异步、削峰的目的。今天这篇,我们来看看SpringBoot是如何集成RabbitMQ,发送消息和消费消息的。同时我们介绍下死信队列。集成RabbitMQ集成RabbitMQ只需要如下... 查看详情
ldap集成rabbitmq
rabbitmq版本:3.7.4rabbitmq支持ldap需要开启插件: rabbitmq-pluginsenablerabbitmq_auth_backend_ldaprabbitmq主要通过两个配置文件对ldap服务进行控制访问:rabbitmq.conf,advanced.config。rabbitmq.conf:ldap连接,查询配置。advanced. 查看详情
RabbitMQ 集成测试和线程
】RabbitMQ集成测试和线程【英文标题】:RabbitMQIntegrationTestandThreading【发布时间】:2015-03-2120:35:24【问题描述】:我通过实现MessageListener接口并设置SimpleMessageListenerContainer编写了一个RabbitMQ消费者。当我手动测试它时它运行良好。... 查看详情
springboot入门:集成rabbitmq消息队列
本片文章续《SpringBoot入门(七):集成swagger2》,关于RabbitMQ的介绍请参考《java基础(六):RabbitMQ入门》1.增加依赖1<!--rabbitMq-->2<dependency>3<groupId>org.springframework.boot</groupId>4<artifactId>spring 查看详情
rabbitmq入门springboot集成rabbitmq(代码片段)
前言本文主要参考了UnderstandingAMQP,theprotocolusedbyRabbitMQ要学习rabbitmq,就要先学习amqp协议,amqp全称AdvancedMessageQueuingProtocol(AMQP),rabbitmq实现了amqp,当然rocketmq也实现了amqp。amqp中有几个重要的概念ÿ 查看详情
springboot集成rabbitmq
...小王叔叔的博客欢迎来访支持:点赞收藏关注RabbitMQ即一个消息队列,主要是用来实现应用程序的异步和解耦,同时也能起到消息缓冲,消息分发的作用。1.RabbitMQ介绍RabbitMQ是实现AMQP(高级消息队列协议)的消息中间... 查看详情
springboot集成rabbitmq
一、SpringBoot集成RabbitMQ非常简单,直接引入配置好的starter依赖,然后配置相关的信息就可以直接使用了。1、引入依赖<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId& 查看详情
spring集成rabbitmq
mq.propertiesmq.host=主机ipmq.username=adminmq.password=admin123mq.port=5672mq.queue.vip=test-queuemq.exchange=test-exchangemq.vhost=testspring-rabbitmq.xml<?xmlversion="1.0"encoding="UTF-8"?>< 查看详情
多语言堆栈的集成测试(Java/MongoDB/RabbitMQ...)
】多语言堆栈的集成测试(Java/MongoDB/RabbitMQ...)【英文标题】:Integrationtestsofapolyglotstack(Java/MongoDB/RabbitMQ...)【发布时间】:2013-02-1816:31:08【问题描述】:我知道RabbitMQ是用Erlang编写的,因此不能像我们使用ActiveMQJMS代理那样嵌入... 查看详情
springboot整合二集成rabbitmq
1、在application.yml文件中进行RabbitMQ的相关配置先上代码spring:rabbitmq:host:192168.21.11port:5672username:guestpassword:passwordpublisher-confirms:true#消息发送到交换机确认机制,是否确认回调 2.项目启动配置 查看详情
springboot集成rabbitmq
1、application.yml配置spring:rabbitmq:host:localhostport:5672listener:simple:acknowledge-mode:manual//手动签发prefetch:1retry:enabled:trueinitial-interval:60000#第一次和第二次尝试发布或传递消息间隔:1分钟max-attempts:8max-interv 查看详情
rabbitmq——springboot集成rabbitmq(代码片段)
...gBoot工程——消息接收者3.测试结果3.1direct3.2fanout3.3topic3.4RabbitMQ管控台中查看SpringBoot工程创建的交换机和消息队列1.创建一个SpringBoot工程——消息发送者前两步都是一样的,只不过在依赖项页面中,要勾选Rabb 查看详情
rabbitmq安装和使用(和spring集成)
一、安装RabbitMQ RabbitMQ是建立在强大的ErlangOTP平台上,因此安装RabbitMQ的前提是安装Erlang。通过下面两个连接下载安装3.2.3版本:下载并安装 EralngOTPForWindows (vR16B03)运行安装 RabbitMQServerWindowsInstaller (v3.2.3)具... 查看详情
rabbitmq集成springboot消息队列入门demo
springboot集成RabbitMq还是很方便的。现在来一个简单的例子来集成rabbitmq。入门demo。主要概念:其中比较重要的概念有4个,分别为:虚拟主机,交换机,队列,和绑定。虚拟主机:一个虚拟主机持有一组交换机、队列和绑定。为... 查看详情