二丶SpringBoot整合RabbitMQ
什么是RabbitMQ?
rabbitMQ是一个开源的消息代理和队列的服务器,用来通过普通的协议在完全不同的应用之前共享数据,rabbitmq是Erlang语言编写的,并且rabbitMQ是基于AMQP协议的
1.创建product
<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.1.5.RELEASE</version> <relativePath/> <!-- lookup parent from repository --></parent><properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- springboot rabbitmq(amqp) -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>
server.servlet.context-path=/server.port=8001spring.rabbitmq.addresses=192.168.12.128:5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=15000
## 使用启用消息确认模式
spring.rabbitmq.publisher-confirms=true
## 设置return消息模式,注意要和mandatory一起去配合使用
##spring.rabbitmq.publisher-returns=true
##spring.rabbitmq.template.mandatory=true
spring.application.name=rabbit-producer
spring.http.encoding.charset=UTF-8
spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
spring.jackson.time-zone=GMT+8
spring.jackson.default-property-inclusion=NON_NULL
@SpringBootApplicationpublic class Application {public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}
package com.bfxy.rabbit.producer.component;import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
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;
import java.util.UUID;
/**
* @auther daiison
* @create 2020-04-2020/4/26-10:30 上午
**/
@Component
public class RabbitSender {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 这里是确认消息的回调监听接口
* 用于确认消息是否被broker所收到
*/
final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
/**
*
* @param correlationData 作为唯一的标识
* @param b
* @param s
* ack brober 是否罗盘成功
* cause 失败一些异常信息
*/
@Override
public void confirm(CorrelationData correlationData, boolean b, String s) {
System.err.println("ack结果为:" + b+","+correlationData);
}
};
/**
* 对外发送消息的方法
*
* @param message 具体的消息
* @param properties 附加额外的消息
* @throws Exception
*/
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);
// 唯一的id
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString().trim());
//发送消息
rabbitTemplate.convertAndSend("exchange-1",
"springboot.rabbit",
msg,
new MessagePostProcessor() {
@Override
public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
System.err.println("--->post to do:" + message);
return message;
}
},
correlationData);
}
}
test测试
package com.bfxy.rabbit.producer;import com.bfxy.rabbit.producer.component.RabbitSender;
import org.junit.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.awt.*;
import java.util.HashMap;
import java.util.Map;
/**
* @auther daiison
* @create 2020-04-2020/4/26-11:48 上午
**/
@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTest {
@Autowired
private RabbitSender rabbitSender;
@Test
public void testSender() throws Exception {
Map<String, Object> properties = new HashMap<String, Object>();
properties.put("attr1", "12345");
properties.put("attr2", "abcde");
rabbitSender.send("hello rabbitmq", properties);
Thread.sleep(10000);
}
}
2. 创建consumer
<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.1.5.RELEASE</version> <relativePath/> <!-- lookup parent from repository --></parent><properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- springboot rabbitmq(amqp) -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>
</project>
server.servlet.context-path=/server.port=8002spring.rabbitmq.addresses=192.168.12.128:5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
spring.rabbitmq.connection-timeout=15000
## 使用启用消息确认模式
## 表示消费者消费成功以后需要手工进行签收(ack) 默认为auto
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10
spring.rabbitmq.listener.simple.prefetch=1
## 设置return消息模式,注意要和mandatory一起去配合使用
##spring.rabbitmq.publisher-returns=true
##spring.rabbitmq.template.mandatory=true
spring.application.name=rabbit-producer
spring.http.encoding.charset=UTF-8
spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
spring.jackson.time-zone=GMT+8
spring.jackson.default-property-inclusion=NON_NULL
@SpringBootApplicationpublic class Application {public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}
package com.bfxy.rabbit.consumer.component;import org.springframework.messaging.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;
/**
* @auther daiison
* @create 2020-04-2020/4/26-11:08 上午
**/
@Component
public class RabbitReceive {
/**
* 组合使用,消息监听
*
* @param message
* @param channel
* @throws Exception
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "queue-1", durable = "true"),
exchange = @Exchange(name = "exchange-1", durable = "true", type = "topic", ignoreDeclarationExceptions = "true"),
key = "springboot.*"
)
)
@RabbitHandler
public void onMessage(Message message, Channel channel) throws Exception {
//1.收到消息之后进行业务端消费处理
System.err.println("----------------");
System.err.println("----- 消费消息------:" + message.getPayload());
//2.处理成功之后,获取deliveryTag 进行手动的ack 签收处理
// application 配置的手动签收模式
Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
//手动签收 ack
channel.basicAck(deliveryTag, false);
}
}
以上是 二丶SpringBoot整合RabbitMQ 的全部内容, 来源链接: utcz.com/z/515891.html