@ -0,0 +1,14 @@
|
|||||||
|
package com.baiye.demo.case25.compensation;
|
||||||
|
|
||||||
|
import org.springframework.boot.SpringApplication;
|
||||||
|
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||||
|
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||||
|
|
||||||
|
@SpringBootApplication
|
||||||
|
@EnableScheduling
|
||||||
|
public class CommonMistakesApplication {
|
||||||
|
public static void main(String[] args) {
|
||||||
|
SpringApplication.run(CommonMistakesApplication.class, args);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,37 @@
|
|||||||
|
package com.baiye.demo.case25.compensation;
|
||||||
|
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
|
@Component
|
||||||
|
@Slf4j
|
||||||
|
public class MemberService {
|
||||||
|
/**
|
||||||
|
* 发送欢迎消息的状态
|
||||||
|
*/
|
||||||
|
private Map<Long, Boolean> welcomeStatus = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
|
//监听用户注册成功的消息,发送欢迎消息
|
||||||
|
@RabbitListener(queues = RabbitConfiguration.QUEUE)
|
||||||
|
public void listen(User user) {
|
||||||
|
log.info("receive mq user {}", user.getId());
|
||||||
|
welcome(user);
|
||||||
|
}
|
||||||
|
|
||||||
|
// 发送欢迎消息
|
||||||
|
public void welcome(User user) {
|
||||||
|
// 去重操作
|
||||||
|
if (welcomeStatus.putIfAbsent(user.getId(), true) == null) {
|
||||||
|
try {
|
||||||
|
TimeUnit.SECONDS.sleep(2);
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
}
|
||||||
|
log.info("memberService: welcome new user {}", user.getId());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,31 @@
|
|||||||
|
package com.baiye.demo.case25.compensation;
|
||||||
|
|
||||||
|
import org.springframework.amqp.core.*;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
|
||||||
|
@Configuration
|
||||||
|
public class RabbitConfiguration {
|
||||||
|
|
||||||
|
public static final String QUEUE = "newuserQueueCompensation";
|
||||||
|
public static final String EXCHANGE = "newuserExchangeCompensation";
|
||||||
|
public static final String ROUTING_KEY = "newuserRoutingCompensation";
|
||||||
|
|
||||||
|
//队列
|
||||||
|
@Bean
|
||||||
|
public Queue queue() {
|
||||||
|
return new Queue(QUEUE);
|
||||||
|
}
|
||||||
|
|
||||||
|
//交换器
|
||||||
|
@Bean
|
||||||
|
public Exchange exchange() {
|
||||||
|
return ExchangeBuilder.directExchange(EXCHANGE).durable(true).build();
|
||||||
|
}
|
||||||
|
|
||||||
|
//绑定
|
||||||
|
@Bean
|
||||||
|
public Binding binding() {
|
||||||
|
return BindingBuilder.bind(queue()).to(exchange()).with(ROUTING_KEY).noargs();
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,16 @@
|
|||||||
|
package com.baiye.demo.case25.compensation;
|
||||||
|
|
||||||
|
import lombok.Data;
|
||||||
|
|
||||||
|
import java.io.Serializable;
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.UUID;
|
||||||
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
|
|
||||||
|
@Data
|
||||||
|
public class User implements Serializable {
|
||||||
|
private static AtomicLong atomicLonng = new AtomicLong();
|
||||||
|
private Long id = atomicLonng.incrementAndGet();
|
||||||
|
private String name = UUID.randomUUID().toString();
|
||||||
|
private LocalDateTime registerTime = LocalDateTime.now();
|
||||||
|
}
|
@ -0,0 +1,36 @@
|
|||||||
|
package com.baiye.demo.case25.compensation;
|
||||||
|
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RequestMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
|
||||||
|
import java.util.concurrent.ThreadLocalRandom;
|
||||||
|
import java.util.stream.IntStream;
|
||||||
|
|
||||||
|
@RestController
|
||||||
|
@Slf4j
|
||||||
|
@RequestMapping("user")
|
||||||
|
public class UserController {
|
||||||
|
@Autowired
|
||||||
|
private UserService userService;
|
||||||
|
@Autowired
|
||||||
|
private RabbitTemplate rabbitTemplate;
|
||||||
|
|
||||||
|
@GetMapping("register")
|
||||||
|
public void register() {
|
||||||
|
// 模拟10个用户注册
|
||||||
|
IntStream.rangeClosed(1, 10).forEach(i -> {
|
||||||
|
//落库
|
||||||
|
User user = userService.register();
|
||||||
|
// 模拟50%的消息可能发送失败
|
||||||
|
if (ThreadLocalRandom.current().nextInt(10) % 2 == 0) {
|
||||||
|
// 通过RabbitMQ发送消息
|
||||||
|
rabbitTemplate.convertAndSend(RabbitConfiguration.EXCHANGE, RabbitConfiguration.ROUTING_KEY, user);
|
||||||
|
log.info("sent mq user {}", user.getId());
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,25 @@
|
|||||||
|
package com.baiye.demo.case25.compensation;
|
||||||
|
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
|
@Service
|
||||||
|
public class UserService {
|
||||||
|
private List<User> users = new ArrayList<>();
|
||||||
|
|
||||||
|
public User register() {
|
||||||
|
User user = new User();
|
||||||
|
users.add(user);
|
||||||
|
return user;
|
||||||
|
}
|
||||||
|
|
||||||
|
public List<User> getUsersAfterIdWithLimit(long id, int limit) {
|
||||||
|
return users.stream()
|
||||||
|
.filter(user -> user.getId() >= id)
|
||||||
|
.limit(limit)
|
||||||
|
.collect(Collectors.toList());
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,12 @@
|
|||||||
|
package com.baiye.demo.case25.deadletter;
|
||||||
|
|
||||||
|
import org.springframework.boot.SpringApplication;
|
||||||
|
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||||
|
|
||||||
|
@SpringBootApplication
|
||||||
|
public class CommonMistakesApplication {
|
||||||
|
public static void main(String[] args) {
|
||||||
|
SpringApplication.run(CommonMistakesApplication.class, args);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,13 @@
|
|||||||
|
package com.baiye.demo.case25.deadletter;
|
||||||
|
|
||||||
|
public class Consts {
|
||||||
|
public static final String QUEUE = "test";
|
||||||
|
public static final String EXCHANGE = "test";
|
||||||
|
public static final String ROUTING_KEY = "test";
|
||||||
|
|
||||||
|
public static final String DEAD_EXCHANGE = "deadtest";
|
||||||
|
public static final String DEAD_QUEUE = "deadtest";
|
||||||
|
public static final String DEAD_ROUTING_KEY = "deadtest";
|
||||||
|
|
||||||
|
|
||||||
|
}
|
@ -0,0 +1,31 @@
|
|||||||
|
package com.baiye.demo.case25.deadletter;
|
||||||
|
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RequestMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
|
||||||
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
|
|
||||||
|
@RequestMapping("deadletter")
|
||||||
|
@Slf4j
|
||||||
|
@RestController
|
||||||
|
public class DeadLetterController {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 自增消息标识
|
||||||
|
*/
|
||||||
|
AtomicLong atomicLong = new AtomicLong();
|
||||||
|
@Autowired
|
||||||
|
private RabbitTemplate rabbitTemplate;
|
||||||
|
|
||||||
|
@GetMapping("sendMessage")
|
||||||
|
public void sendMessage() {
|
||||||
|
String msg = "msg" + atomicLong.incrementAndGet();
|
||||||
|
log.info("send message {}", msg);
|
||||||
|
// 发送消息
|
||||||
|
rabbitTemplate.convertAndSend(Consts.EXCHANGE, msg);
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,27 @@
|
|||||||
|
package com.baiye.demo.case25.deadletter;
|
||||||
|
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
@Component
|
||||||
|
@Slf4j
|
||||||
|
public class MQListener {
|
||||||
|
|
||||||
|
|
||||||
|
@RabbitListener(queues = Consts.QUEUE)
|
||||||
|
public void handler(String data) {
|
||||||
|
//http://localhost:15672/#/
|
||||||
|
log.info("got message {}", data);
|
||||||
|
throw new NullPointerException("error");
|
||||||
|
//throw new AmqpRejectAndDontRequeueException("error");
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 死信队列处理程序
|
||||||
|
*/
|
||||||
|
@RabbitListener(queues = Consts.DEAD_QUEUE)
|
||||||
|
public void deadHandler(String data) {
|
||||||
|
log.error("got dead message {}", data);
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,12 @@
|
|||||||
|
package com.baiye.demo.case25.fanoutvswork;
|
||||||
|
|
||||||
|
import org.springframework.boot.SpringApplication;
|
||||||
|
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||||
|
|
||||||
|
@SpringBootApplication
|
||||||
|
public class CommonMistakesApplication {
|
||||||
|
public static void main(String[] args) {
|
||||||
|
SpringApplication.run(CommonMistakesApplication.class, args);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,67 @@
|
|||||||
|
package com.baiye.demo.case25.fanoutvswork;
|
||||||
|
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.amqp.core.BindingBuilder;
|
||||||
|
import org.springframework.amqp.core.Declarables;
|
||||||
|
import org.springframework.amqp.core.FanoutExchange;
|
||||||
|
import org.springframework.amqp.core.Queue;
|
||||||
|
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||||
|
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RequestMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
|
||||||
|
import java.util.UUID;
|
||||||
|
|
||||||
|
@Slf4j
|
||||||
|
@Configuration
|
||||||
|
@RestController
|
||||||
|
@RequestMapping("fanoutright")
|
||||||
|
public class FanoutQueueRight {
|
||||||
|
private static final String MEMBER_QUEUE = "newusermember";
|
||||||
|
private static final String PROMOTION_QUEUE = "newuserpromotion";
|
||||||
|
private static final String EXCHANGE = "newuser";
|
||||||
|
@Autowired
|
||||||
|
private RabbitTemplate rabbitTemplate;
|
||||||
|
|
||||||
|
@GetMapping
|
||||||
|
public void sendMessage() {
|
||||||
|
rabbitTemplate.convertAndSend(EXCHANGE, "", UUID.randomUUID().toString());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public Declarables declarables() {
|
||||||
|
Queue memberQueue = new Queue(MEMBER_QUEUE);
|
||||||
|
Queue promotionQueue = new Queue(PROMOTION_QUEUE);
|
||||||
|
|
||||||
|
FanoutExchange exchange = new FanoutExchange(EXCHANGE);
|
||||||
|
return new Declarables(memberQueue, promotionQueue, exchange,
|
||||||
|
BindingBuilder.bind(memberQueue).to(exchange),
|
||||||
|
BindingBuilder.bind(promotionQueue).to(exchange));
|
||||||
|
}
|
||||||
|
|
||||||
|
@RabbitListener(queues = MEMBER_QUEUE)
|
||||||
|
public void memberService1(String userName) {
|
||||||
|
log.info("memberService1: welcome message sent to new user {}", userName);
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
@RabbitListener(queues = MEMBER_QUEUE)
|
||||||
|
public void memberService2(String userName) {
|
||||||
|
log.info("memberService2: welcome message sent to new user {}", userName);
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
@RabbitListener(queues = PROMOTION_QUEUE)
|
||||||
|
public void promotionService1(String userName) {
|
||||||
|
log.info("promotionService1: gift sent to new user {}", userName);
|
||||||
|
}
|
||||||
|
|
||||||
|
@RabbitListener(queues = PROMOTION_QUEUE)
|
||||||
|
public void promotionService2(String userName) {
|
||||||
|
log.info("promotionService2: gift sent to new user {}", userName);
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,51 @@
|
|||||||
|
package com.baiye.demo.case25.fanoutvswork;
|
||||||
|
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.amqp.core.BindingBuilder;
|
||||||
|
import org.springframework.amqp.core.Declarables;
|
||||||
|
import org.springframework.amqp.core.DirectExchange;
|
||||||
|
import org.springframework.amqp.core.Queue;
|
||||||
|
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||||
|
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RequestMapping;
|
||||||
|
|
||||||
|
import java.util.UUID;
|
||||||
|
|
||||||
|
@Slf4j
|
||||||
|
//@Configuration
|
||||||
|
//@RestController
|
||||||
|
@RequestMapping("workqueueright")
|
||||||
|
public class WorkQueueRight {
|
||||||
|
|
||||||
|
private static final String EXCHANGE = "newuserExchange";
|
||||||
|
private static final String QUEUE = "newuserQueue";
|
||||||
|
@Autowired
|
||||||
|
private RabbitTemplate rabbitTemplate;
|
||||||
|
|
||||||
|
@GetMapping
|
||||||
|
public void sendMessage() {
|
||||||
|
rabbitTemplate.convertAndSend(EXCHANGE, "test", UUID.randomUUID().toString());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public Queue queue() {
|
||||||
|
return new Queue(QUEUE);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public Declarables declarables() {
|
||||||
|
DirectExchange exchange = new DirectExchange(EXCHANGE);
|
||||||
|
return new Declarables(queue(), exchange,
|
||||||
|
BindingBuilder.bind(queue()).to(exchange).with("test"));
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
@RabbitListener(queues = "#{queue.name}")
|
||||||
|
public void memberService(String userName) {
|
||||||
|
log.info("memberService: welcome message sent to new user {}", userName);
|
||||||
|
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,12 @@
|
|||||||
|
package com.baiye.demo.case25.rabbitmqdlx;
|
||||||
|
|
||||||
|
import org.springframework.boot.SpringApplication;
|
||||||
|
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||||
|
|
||||||
|
@SpringBootApplication
|
||||||
|
public class CommonMistakesApplication {
|
||||||
|
public static void main(String[] args) {
|
||||||
|
SpringApplication.run(CommonMistakesApplication.class, args);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,19 @@
|
|||||||
|
package com.baiye.demo.case25.rabbitmqdlx;
|
||||||
|
|
||||||
|
public class Consts {
|
||||||
|
public static final Integer RETRY_INTERNAL = 3000;
|
||||||
|
public static final Integer RETRY_COUNT = 2;
|
||||||
|
|
||||||
|
|
||||||
|
public static final String EXCHANGE = "worker";
|
||||||
|
public static final String QUEUE = "worker";
|
||||||
|
public static final String ROUTING_KEY = "worker";
|
||||||
|
|
||||||
|
public static final String BUFFER_QUEUE = "buffer";
|
||||||
|
public static final String BUFFER_EXCHANGE = "buffer";
|
||||||
|
public static final String BUFFER_ROUTING_KEY = "buffer";
|
||||||
|
|
||||||
|
public static final String DEAD_EXCHANGE = "dead";
|
||||||
|
public static final String DEAD_QUEUE = "dead";
|
||||||
|
public static final String DEAD_ROUTING_KEY = "dead";
|
||||||
|
}
|
@ -0,0 +1,27 @@
|
|||||||
|
package com.baiye.demo.case25.rabbitmqdlx;
|
||||||
|
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RequestMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
|
||||||
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
|
|
||||||
|
@RequestMapping("deadletter")
|
||||||
|
@Slf4j
|
||||||
|
@RestController
|
||||||
|
public class DeadLetterController {
|
||||||
|
|
||||||
|
AtomicLong atomicLong = new AtomicLong();
|
||||||
|
@Autowired
|
||||||
|
private RabbitTemplate rabbitTemplate;
|
||||||
|
|
||||||
|
@GetMapping("send")
|
||||||
|
public void send() {
|
||||||
|
String message = "msg" + atomicLong.incrementAndGet();
|
||||||
|
log.info("Client 发送消息 {}", message);
|
||||||
|
rabbitTemplate.convertAndSend(Consts.EXCHANGE, Consts.QUEUE, message);
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,64 @@
|
|||||||
|
package com.baiye.demo.case25.rabbitmqdlx;
|
||||||
|
|
||||||
|
import com.rabbitmq.client.AMQP;
|
||||||
|
import com.rabbitmq.client.Channel;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.amqp.core.Message;
|
||||||
|
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||||
|
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.messaging.handler.annotation.Payload;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
|
|
||||||
|
@Component
|
||||||
|
@Slf4j
|
||||||
|
public class MQListener {
|
||||||
|
@Autowired
|
||||||
|
private MessagePropertiesConverter messagePropertiesConverter;
|
||||||
|
|
||||||
|
@RabbitListener(queues = Consts.QUEUE)
|
||||||
|
public void handler(@Payload Message message, Channel channel) throws IOException {
|
||||||
|
String m = new String(message.getBody());
|
||||||
|
try {
|
||||||
|
log.info("Handler 收到消息:{}", m);
|
||||||
|
throw new RuntimeException("处理消息失败");
|
||||||
|
} catch (Exception e) {
|
||||||
|
Map<String, Object> headers = message.getMessageProperties().getHeaders();
|
||||||
|
Long retryCount = getRetryCount(headers);
|
||||||
|
if (retryCount < Consts.RETRY_COUNT) {
|
||||||
|
log.info("Handler 消费消息:{} 异常,准备重试第{}次", m, ++retryCount);
|
||||||
|
|
||||||
|
AMQP.BasicProperties rabbitMQProperties =
|
||||||
|
messagePropertiesConverter.fromMessageProperties(message.getMessageProperties(), "UTF-8");
|
||||||
|
rabbitMQProperties.builder().headers(headers);
|
||||||
|
channel.basicPublish(Consts.BUFFER_EXCHANGE, Consts.BUFFER_ROUTING_KEY, rabbitMQProperties, message.getBody());
|
||||||
|
} else {
|
||||||
|
log.info("Handler 消费消息:{} 异常,已重试 {} 次,发送到死信队列处理!", m, Consts.RETRY_COUNT);
|
||||||
|
channel.basicPublish(Consts.DEAD_EXCHANGE, Consts.DEAD_ROUTING_KEY, null, message.getBody());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private long getRetryCount(Map<String, Object> headers) {
|
||||||
|
long retryCount = 0;
|
||||||
|
if (null != headers) {
|
||||||
|
if (headers.containsKey("x-death")) {
|
||||||
|
List<Map<String, Object>> deathList = (List<Map<String, Object>>) headers.get("x-death");
|
||||||
|
if (!deathList.isEmpty()) {
|
||||||
|
Map<String, Object> deathEntry = deathList.get(0);
|
||||||
|
retryCount = (Long) deathEntry.get("count");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return retryCount;
|
||||||
|
}
|
||||||
|
|
||||||
|
@RabbitListener(queues = Consts.DEAD_QUEUE)
|
||||||
|
public void deadHandler(@Payload Message message) {
|
||||||
|
log.error("DeadHandler 收到死信消息: {}", new String(message.getBody()));
|
||||||
|
}
|
||||||
|
}
|
@ -0,0 +1,68 @@
|
|||||||
|
package com.baiye.demo.case25.rabbitmqdlx;
|
||||||
|
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.amqp.core.*;
|
||||||
|
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
|
||||||
|
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
|
||||||
|
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
|
||||||
|
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
|
||||||
|
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
|
||||||
|
import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
|
||||||
|
|
||||||
|
@Configuration
|
||||||
|
@Slf4j
|
||||||
|
public class RabbitConfiguration implements RabbitListenerConfigurer {
|
||||||
|
@Override
|
||||||
|
public void configureRabbitListeners(RabbitListenerEndpointRegistrar rabbitListenerEndpointRegistrar) {
|
||||||
|
rabbitListenerEndpointRegistrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public MessageHandlerMethodFactory messageHandlerMethodFactory() {
|
||||||
|
DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory();
|
||||||
|
messageHandlerMethodFactory.setMessageConverter(consumerJackson2MessageConverter());
|
||||||
|
return messageHandlerMethodFactory;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public MappingJackson2MessageConverter consumerJackson2MessageConverter() {
|
||||||
|
return new MappingJackson2MessageConverter();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public MessagePropertiesConverter messagePropertiesConverter() {
|
||||||
|
return new DefaultMessagePropertiesConverter();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public Declarables declarablesForWorker() {
|
||||||
|
Queue queue = new Queue(Consts.QUEUE);
|
||||||
|
DirectExchange directExchange = new DirectExchange(Consts.EXCHANGE);
|
||||||
|
return new Declarables(queue, directExchange,
|
||||||
|
BindingBuilder.bind(queue).to(directExchange).with(Consts.ROUTING_KEY));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public Declarables declarablesForBuffer() {
|
||||||
|
Queue queue = QueueBuilder.durable(Consts.BUFFER_QUEUE)
|
||||||
|
.withArgument("x-dead-letter-exchange", Consts.EXCHANGE)
|
||||||
|
.withArgument("x-dead-letter-routing-key", Consts.ROUTING_KEY)
|
||||||
|
.withArgument("x-message-ttl", Consts.RETRY_INTERNAL)
|
||||||
|
.build();
|
||||||
|
DirectExchange directExchange = new DirectExchange(Consts.BUFFER_EXCHANGE);
|
||||||
|
return new Declarables(queue, directExchange,
|
||||||
|
BindingBuilder.bind(queue).to(directExchange).with(Consts.BUFFER_ROUTING_KEY));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public Declarables declarablesForDead() {
|
||||||
|
Queue queue = new Queue(Consts.DEAD_QUEUE);
|
||||||
|
DirectExchange directExchange = new DirectExchange(Consts.DEAD_EXCHANGE);
|
||||||
|
return new Declarables(queue, directExchange,
|
||||||
|
BindingBuilder.bind(queue).to(directExchange).with(Consts.DEAD_ROUTING_KEY));
|
||||||
|
}
|
||||||
|
}
|
After Width: | Height: | Size: 58 KiB |
After Width: | Height: | Size: 35 KiB |
After Width: | Height: | Size: 33 KiB |
After Width: | Height: | Size: 27 KiB |
After Width: | Height: | Size: 23 KiB |
After Width: | Height: | Size: 29 KiB |
After Width: | Height: | Size: 87 KiB |
After Width: | Height: | Size: 46 KiB |