SpringBoot集成Rabbit使用TopicRabbit指定发送集合

Rabbitmq中绑定

exchange:flow

routing-key:user

bind-queue:flow_user

白话文就是,把user绑定到flow_user序列

发送方使用routing-key推送：

//把routing-key发送给名为flow的exchenge，然后exchenge负责向绑定的这个Queue推送
 amqpTemplate.convertAndSend("flow","user", context);

Rabbit配置

• 添加exchange（这里类型type应该是topic,截图时候没有注意）

添加exchange

• 添加Queue

添加Queue

• 添加这个User 到exchange(注意routing-key)

SpringBoot集成Rabbitmq

• 注册配置bean

@Configurable
public class TopicRabbitConfig {
    public final static String FLOW = "flow";
    
    public final static String USER = "user";
    public final static String USER_QUEUE = "flow_user";

   
    @Bean
    public Queue queueMessages3() {
        return new Queue(USER_QUEUE);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(FLOW);
    }
    @Bean
    Binding bindingExchangeMessages3(Queue queueMessages3, TopicExchange exchange) {
        return BindingBuilder.bind(queueMessages3).to(exchange).with(FLOW);
    }
}

• 发送方代码

/**
 * @Package: pterosaur.account.service.impl
 * @Description: 模拟发送消息,测试使用
 * @author: liuxin
 * @date: 17/4/19 下午3:17
 */
@Component
public class AccountSentImpl {
    @Autowired
    private AmqpTemplate amqpTemplate;

    private ExecutorService threadPool = Executors.newFixedThreadPool(8);

    public void send() {
       for (int i=0;i<10;i++){
           String context = "hello :" + DateUtil.formatDatetime(System.currentTimeMillis())+",当前线程:"+Thread.currentThread().getName();
           System.out.println("Sender : " + context);
           threadPool.execute(new Runnable() {
               @Override
               public void run() {
                   amqpTemplate.convertAndSend(TopicRabbitConfig.FLOW,TopicRabbitConfig.USER, context);
               }
           });
       }
    }

}

• 接受方代码

/**
 * @Package: pterosaur.account.service.impl
 * @Description: mq信息处理实现类
 * @author: liuxin
 * @date: 17/4/19 下午2:55
 */
@Component
public class AccountReceiverImpl implements AccountReceiver {
    private static final Logger logger = LoggerFactory.getLogger(AccountReceiverImpl.class);

    @Autowired
    ExecutorService threadPool;


    /**
     * 用户流水
     *
     * @param message
     */
    @RabbitListener(queues = TopicRabbitConfig.USER_QUEUE)
    @RabbitHandler
    public void processUser(String message) {
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                logger.info("用户侧流水:{}",message);
            }
        });
    }


}

• 测试代码

Sender : hello :2017-04-25 17:44:15,当前线程:main
Sender : hello :2017-04-25 17:44:20,当前线程:main
2017-04-25 17:44:25.754  INFO 67685 --- [pool-1-thread-1] p.a.service.impl.AccountReceiverImpl     : 用户侧流水:hello :2017-04-25 17:44:20,当前线程:main
Sender : hello :2017-04-25 17:44:25,当前线程:main
Sender : hello :2017-04-25 17:44:30,当前线程:main
2017-04-25 17:44:32.048  INFO 67685 --- [pool-1-thread-2] p.a.service.impl.AccountReceiverImpl     : 用户侧流水:hello :2017-04-25 17:44:30,当前线程:main
Sender : hello :2017-04-25 17:44:32,当前线程:main
Sender : hello :2017-04-25 17:44:33,当前线程:main
2017-04-25 17:44:35.556  INFO 67685 --- [pool-1-thread-3] p.a.service.impl.AccountReceiverImpl     : 用户侧流水:hello :2017-04-25 17:44:33,当前线程:main
Sender : hello :2017-04-25 17:44:35,当前线程:main
Sender : hello :2017-04-25 17:44:37,当前线程:main
2017-04-25 17:44:38.797  INFO 67685 --- [pool-1-thread-1] p.a.service.impl.AccountReceiverImpl     : 用户侧流水:hello :2017-04-25 17:44:37,当前线程:main