pom.xml
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.rabbit</groupId>
<artifactId>rabbit-producer</artifactId>
<version>0.0.1-SNAPSHOT</version>
<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>
application.properties
server.servlet.context-path=/
server.port=8002
spring.rabbitmq.addresses=192.168.11.71:5672,192.168.11.72:5672,192.168.11.71:5673
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
## 作业:
## 最好不要在代码里写死配置信息,尽量使用这种方式也就是配置文件的方式
## 在代码里使用 ${} 方式进行设置配置: ${spring.rabbitmq.listener.order.exchange.name}
#spring.rabbitmq.listener.order.exchange.name=order-exchange
#spring.rabbitmq.listener.order.exchange.durable=true
#spring.rabbitmq.listener.order.exchange.type=topic
#spring.rabbitmq.listener.order.exchange.key=order.*
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
RabbitReceive.java
package com.bfxy.rabbit.consumer.component;
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.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;
@Component
public class RabbitReceive {
/**
* 组合使用监听
* @RabbitListener @QueueBinding @Queue @Exchange
* @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操作, 因为我们配置文件里配置的是 手工签收
// spring.rabbitmq.listener.simple.acknowledge-mode=manual
Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
channel.basicAck(deliveryTag, false);
}
}
RabbitSender.java
package com.bfxy.rabbit.producer.component;
import java.util.Map;
import java.util.UUID;
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.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
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;
@Component
public class RabbitSender {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 这里就是确认消息的回调监听接口,用于确认消息是否被broker所收到
*/
final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
/**
* @param CorrelationData 作为一个唯一的标识
* @param ack broker 是否落盘成功
* @param cause 失败的一些异常信息
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
System.err.println("消息ACK结果:" + ack + ", correlationData: " + correlationData.getId());
}
};
/**
* 对外发送消息的方法
* @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());
MessagePostProcessor mpp = 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;
}
};
rabbitTemplate.convertAndSend("exchange-1",
"springboot.rabbit",
msg, mpp, correlationData);
}
}
ApplicationTests.java
package com.bfxy.rabbit.producer.test;
import java.util.HashMap;
import java.util.Map;
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 com.bfxy.rabbit.producer.component.RabbitSender;
@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests {
@Autowired
private RabbitSender reRabbitSender;
@Test
public void testSender() throws Exception {
Map<String, Object> properties = new HashMap<String, Object>();
properties.put("attr1", "12345");
properties.put("attr2", "abcde");
reRabbitSender.send("hello rabbitmq!", properties);
Thread.sleep(10000);
}
}
Application.java
package com.bfxy.rabbit.consumer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}
consumer采用spring boot app启动!!!
最后修改于 2020-07-14 09:59:10
如果觉得我的文章对你有用,请随意赞赏
扫一扫支付

