给注册用户发红包,RabbitMQ实现(分布式事务2)

public interface UserCenterMq {

   /**
    * 用户系统exchange名
    */
   String MQ_EXCHANGE_USER = "user.topic.exchange";

   /**
    * 发送红包routing key
    */
   String ROUTING_KEY_POST_REDPACKET = "post.redpacket";
   //String ROUTING_KEY_POST_REDPACKET = "post.#";
   /**
    * 死信队列：
    */
   String deadRoutingKey = "dead_routing_key";
   String deadExchangeName = "dead_exchange";
}

spring:
  rabbitmq:
      host: 192.168.5.182
      port: 5672
      username: admin
      password: admin
      virtual-host: /
      publisher-confirms: true
      publisher-returns: true
      listener:
        simple:
          acknowledge-mode: manual

@Configuration
public class RabbitmqConfig {

   /**
    * 红包队列名
    */
   public static final String RED_PACKET_QUEUE = "red.packet.queue";
   /**
    * 死信队列：
    */
   public final static String deadQueueName = "dead_queue";
   /**
    * 声明队列，此队列用来接收用户注册的消息
    * 
    * @return
    */
   @Bean
   public Queue redPacketQueue() {
      Queue queue = new Queue(RED_PACKET_QUEUE);
      return queue;
   }
   /**
    * 死信队列：
    */

   @Bean
   public Queue deadQueue() {
      Queue queue = new Queue(deadQueueName, true);
      return queue;
   }

   @Bean
   public DirectExchange deadExchange() {
      return new DirectExchange(deadExchangeName);
   }

   @Bean
   public Binding bindingDeadExchange() {
      return BindingBuilder.bind(deadQueue()).to(deadExchange()).with(deadRoutingKey);
   }

   @Bean
   public Binding bindingRedPacketDeadExchange() {
      return BindingBuilder.bind(redPacketQueue()).to(deadExchange()).with(deadRoutingKey);
   }

   @Bean
   public TopicExchange userTopicExchange() {
      return new TopicExchange(UserCenterMq.MQ_EXCHANGE_USER);
   }

   /**
    * 将红包队列和用户的exchange做个绑定
    * 
    * @return
    */
   @Bean
   public Binding bindingRedPacket() {
      Binding binding = BindingBuilder.bind(redPacketQueue()).to(userTopicExchange())
            .with(UserCenterMq.ROUTING_KEY_POST_REDPACKET);
      return binding;
   }
}

@Component
public class MessageSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send(String exchange, String routingKey,String context) {
        System.out.println("send content = " + context);
        this.rabbitTemplate.setMandatory(true);
        this.rabbitTemplate.setConfirmCallback(this);
        this.rabbitTemplate.setReturnCallback(this);
        this.rabbitTemplate.convertAndSend(exchange, routingKey, context);
    }

    /**
     * 确认后回调:
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack) {
            System.out.println("send ack fail, cause = " + cause);
        } else {
            System.out.println("send ack success");
        }
    }

    /**
     * 失败后return回调：
     *
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("send fail return-message = " + new String(message.getBody()) + ", replyCode: " + replyCode + ", replyText: " + replyText + ", exchange: " + exchange + ", routingKey: " + routingKey);
    }
}

@Component
public class UserTransactionEventListener {
    @Autowired
    private MessageSender messageSender;

    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
    public void beforeCommit(PayloadApplicationEvent<User> event) {
        System.out.println("before commit, id: " + event.getPayload().getId());
    }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void afterCommit(PayloadApplicationEvent<User> event) {
        System.out.println("after commit, id: " + event.getPayload().getId());
    }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMPLETION)
    public void afterCompletion(PayloadApplicationEvent<User> event) {
        System.out.println("after completion, id: " + event.getPayload().getId());
        messageSender.send(UserCenterMq.MQ_EXCHANGE_USER,UserCenterMq.ROUTING_KEY_POST_REDPACKET, JSONObject.toJSONString(event.getPayload()));
        //messageSender.send(UserCenterMq.MQ_EXCHANGE_USER,"post.redpacket", JSONObject.toJSONString(event.getPayload()));
    }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_ROLLBACK)
    public void afterRollback(PayloadApplicationEvent<User> event) {
        System.out.println("after rollback, id: " + event.getPayload().getId());
    }
}

@Data
public class RedPacket implements Serializable {
    private long redPacketId;
    private long userId;
    private Double redPacketAmount;
}

@Mapper
public interface RedPacketDao {
    @Options(useGeneratedKeys = true, keyProperty = "red_packet_id")
    @Insert("insert into red_packet (user_id,red_packet_amount) values (#{userId},#{redPacketAmount})")
    void add(RedPacket redPacket);
}

public interface RedPacketService {
    public void add(RedPacket redPacket);
}

@Transactional
@Service
public class RedPacketServiceImpl implements RedPacketService {
    @Autowired
    private RedPacketDao redPacketDao;
    @Override
    public void add(RedPacket redPacket) {
        redPacketDao.add(redPacket);
    }
}

@Component
@RabbitListener(queues = RabbitmqConfig.RED_PACKET_QUEUE)
public class PostRedPacketConsumer {
    @Autowired
    private RedPacketService redPacketService;

    @RabbitHandler
    public void postRedPacket(String userStr, Channel channel, Message message) throws IOException {
        try {
            //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉；否则消息服务器以为这条消息没处理掉 后续还会在发
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            User user = JSONObject.parseObject(userStr,User.class);
            RedPacket redPacket = new RedPacket();
            redPacket.setUserId(user.getId());
            redPacket.setRedPacketAmount((double)(Math.random() * 10));
            redPacketService.add(redPacket);
        } catch (IOException e) {
            e.printStackTrace();
            //丢弃这条消息
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
            System.out.println("receiver fail");
        }
    }
}