java操作rabbitmq收发消息

//生产者 //配置 !-- 引入rabbitmq-- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency spring: rabbitmq: host: 192.168.200.129 port: 5672 username: guest password: guest # 开启发送端确认 publisher-confirm-type: correlated # 开启发送到队列确认 publisher-returns: true # 异步回调 template: mandatory: true Configuration public class RabbitMQConfig { //使用Jackson转换消息格式为JSON便于调试和传输复杂对象。 Bean public MessageConverter messageConverter() { return new Jackson2JsonMessageConverter(); // } //初始化rabbitmq PostConstruct public void initRabbitTemplate() { //confirm message是否正确发送到交换机 rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { /** * 确认消息是否正确发送到交换机 * param correlationData 消息的唯一标识 * param ack 消息是否正确发送到交换机 * param cause 错误信息 */ Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { System.out.println(confirm...correlationData[correlationData]ack[ack]cause[cause]); } }); rabbitTemplate.setMandatory(true); //消息从exchange发送到queue失败返回处理 rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() { /** * 消息发送失败返回处理 * param message 消息 * param replayCode 错误码 * param replyText 错误信息 * param exchange 交换机 * param routingkey 路由键 */ Override public void returnedMessage(Message message, int replayCode, String replyText, String exchange, String routingkey) { System.out.println(returnedMessage...message[message]replayCode[replayCode]replyText[replyText]exchange[exchange]routingkey[routingkey]); } }); }} //发送消息 rabbitTemplate.convertAndSend(order-event-exchange,order.seckill.order,seckillOrderTo);//消费者 !-- 引入rabbitmq-- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency //配置 spring: rabbitmq: host: 192.168.200.129 port: 5672 username: guest password: guest # 手动确认收到信息 listener: simple: acknowledge-mode: manualConfiguration public class RabbitMQConfig {Bean public MessageConverter messageConverter() { return new Jackson2JsonMessageConverter(); }/** * sec kill queue */ Bean public Queue orderSecKillQueue() { return new Queue(order.seckill.queue, true, false, false); } /** * sec kill binding */ Bean public Binding orderSecKillQueueBinding() { return new Binding(order.seckill.queue, Binding.DestinationType.QUEUE, order-event-exchange, order.seckill.order, null); }} //接收message Slf4j RabbitListener(queues order.seckill.queue) Component public class OrderSeckillListener { Autowired private OrderSeckillService orderSeckillService; RabbitHandler public void listener(SeckillOrderTo to, Channel channel, Message message) { log.info(收到秒杀订单信息{}, to); try { orderSeckillService.createSeckillOrder(to); channel.basicAck(message.getMessageProperties().getDeliveryTag(),false); } catch (Exception e) { throw new RuntimeException(e); } } }