为什么要有红黑树?什么是红黑树?画了20张图,看完这篇你就明白了
一、springboot整合rabbitmq
- 我们需要新建两个工程,一个作为生产者,另一个作为消费者。在pom.xml中添加amqp依赖:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
- 在application.yml文件中添加rabbitmq的相关信息:
spring:
rabbitmq:
# 连接地址
host: 127.0.0.1
# 端口
port: 5672
# 登录账号
username: guest
# 登录密码
password: guest
# 虚拟主机
virtual-host: /
- 在生产者工程中新建配置项rabbitmqConfig.java,申明名称为”byte-zb“直连交换机和队列,使用”byte-zb“的routing-key将队列和交换机绑定,代码如下:
@Configuration
public class RabbitConfig {
public static final String QUEUE_NAME = "byte-zb";
public static final String EXCHANGE_NAME = "byte-zb";
public static final String ROUTING_KEY = "byte-zb";
// 队列申明
@Bean
public Queue queue(){
return new Queue(QUEUE_NAME);
}
// 申明交换机
@Bean
public DirectExchange directExchange(){
return new DirectExchange(EXCHANGE_NAME);
}
// 数据绑定申明
@Bean
public Binding directBinding(){
return BindingBuilder.bind(queue()).to(directExchange()).with(ROUTING_KEY);
}
}
- 创建生产者发送一条消息,代码如下:
@RestController
public class Producer {
public static final String QUEUE_NAME = "byte-zb";
public static final String EXCHANGE_NAME = "byte-zb";
@Autowired
private AmqpTemplate amqpTemplate;
@RequestMapping("/send")
public void sendMessage(){
JSONObject jsonObject = new JSONObject();
jsonObject.put("email","11111111111");
jsonObject.put("timestamp",System.currentTimeMillis());
String json = jsonObject.toJSONString();
System.out.println(json);
amqpTemplate.convertAndSend(EXCHANGE_NAME,QUEUE_NAME,json);
}
}
- 在消费者工程里创建消费者消费消息,代码如下:
@Component
public class Consumer throws Exception{
public static final String QUEUE_NAME = "byte-zb";
@RabbitListener(queues = QUEUE_NAME)
public void receiveMessage(String message){
System.out.println("接收到的消息为"+message);
}
}
我们启动生产者,然后请求send接口,然后打开rabbitmq控制台发现多了一个名为”byte-zb“的交换机和队列,并且队列中出现了一个未消费的消息,然后启动消费者,我们会在控制台上发现打印了一条消息,同时rabbitmq控制台中”byte-zb“的队列中消息没有了。
二、自动补偿机制
如果消费者消息消费不成功的话,会出现什么情况呢?我们修改一下消费者代码,然后看看。
@Component
public class Consumer {
public static final String QUEUE_NAME = "byte-zb";
@RabbitListener(queues = QUEUE_NAME)
public void receiveMessage(String message) throws Exception {
System.out.println("接收到的消息为"+message);
int i = 1 / 0;
}
}
我们会看到消费者工程控制台一直在刷新报错,当消费者配出异常,也就是说当消息消费不成功的话,该消息会存放在rabbitmq的服务端,一直进行重试,直到不抛出异常为止。
如果一直抛异常,我们的服务很容易挂掉,那有没有办法控制重试几次不成功就不再重试了呢?答案是有的。我们在消费者application.yml中增加一段配置。
spring:
rabbitmq:
# 连接地址
host: 127.0.0.1
# 端口
port: 5672
# 登录账号
username: guest
# 登录密码
password: guest
# 虚拟主机
virtual-host: /
listener:
simple:
retry:
enabled: true # 开启消费者进行重试
max-attempts: 5 # 最大重试次数
initial-interval: 3000 # 重试时间间隔
上面配置的意思是消费异常后,重试五次,每次隔3s。继续启动消费者看看效果,我们发现重试五次以后,就不再重试了。
三、结合实际案例来使用消息补偿机制
像上面那种情况出现的异常其实不管怎么重试都不会成功,实际上用到消息补偿的就是调用第三方接口的这种。
案例:生者往队列中扔一条消息,包含邮箱和发送内容。消费者拿到消息后将调用邮件接口发送邮件。有时候可能邮件接口由于网络等原因不通,这时候就需要去重试了。
小白学 Python 爬虫(42):春节去哪里玩(系列终篇)
在调用接口的工具类中,如果出现异常我们直接返回null,工具类具体代码就不贴了,如果返回null之后怎么处理呢?我们只需要抛出异常,rabbitListener捕获到异常后就会自动重试。
我们改造一下消费者代码:
@Component
public class Consumer {
public static final String QUEUE_NAME = "byte-zb";
@RabbitListener(queues = QUEUE_NAME)
public void receiveMessage(String message) throws Exception {
System.out.println("接收到的消息为"+message);
JSONObject jsonObject = JSONObject.parseObject(message);
String email = jsonObject.getString("email");
String content = jsonObject.getString("timestamp");
String httpUrl = "http://127.0.0.1:8080/email?email"+email+"&content="+content;
// 如果发生异常则返回null
String body = HttpUtils.httpGet(httpUrl, "utf-8");
//
if(body == null){
throw new Exception();
}
}
}
当然我们可以自定义异常抛出。具体怎么试验呢,第一步启动生产者和消费者,这时候我们发现消费者在重试,第二步我们启动邮件服务,这时候我们会发现邮件发送成功了,消费者不再重试了。
四、解决消息幂等性问题
一些刚接触java的同学可能对幂等性不太清楚。幂等性就是重复消费造成结果不一致。为了保证幂等性,因此消费者消费消息只能消费一次消息。我么可以是用全局的消息id来控制幂等性。当消息被消费了之后我们可以选择缓存保存这个消息id,然后当再次消费的时候,我们可以查询缓存,如果存在这个消息id,我们就不错处理直接return即可。先改造生产者代码,在消息中添加消息id:
@RequestMapping("/send")
public void sendMessage(){
JSONObject jsonObject = new JSONObject();
jsonObject.put("email","11111111111");
jsonObject.put("timestamp",System.currentTimeMillis());
String json = jsonObject.toJSONString();
System.out.println(json);
Message message = MessageBuilder.withBody(json.getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON)
.setContentEncoding("UTF-8").setMessageId(UUID.randomUUID()+"").build();
amqpTemplate.convertAndSend(EXCHANGE_NAME,QUEUE_NAME,message);
}
消费者代码改造:
@Component
public class Consumer {
public static final String QUEUE_NAME = "byte-zb";
@RabbitListener(queues = QUEUE_NAME)
public void receiveMessage(Message message) throws Exception {
Jedis jedis = new Jedis("localhost", 6379);
String messageId = message.getMessageProperties().getMessageId();
String msg = new String(message.getBody(),"UTF-8");
System.out.println("接收导的消息为:"+msg+"==消息id为:"+messageId);
String messageIdRedis = jedis.get("messageId");
if(messageId == messageIdRedis){
return;
}
JSONObject jsonObject = JSONObject.parseObject(msg);
String email = jsonObject.getString("email");
String content = jsonObject.getString("timestamp");
String httpUrl = "http://127.0.0.1:8080/email?email"+email+"&content="+content;
// 如果发生异常则返回null
String body = HttpUtils.httpGet(httpUrl, "utf-8");
//
if(body == null){
throw new Exception();
}
jedis.set("messageId",messageId);
}
}
我们在消费者端使用redis存储消息id,只做演示,具体项目请根据实际情况选择相应的工具进行存储。
如果文章对您有帮助,请记得点赞关注哟~
欢迎大家关注我的公众号:字节传说,每日推送技术文章供大家学习参考。
自己动手实现智能家居之温湿度数码管展示(四位共阳数码管)