@Test @SneakyThrows public void process() { Thread.sleep(1 * 1000); Executors.newSingleThreadScheduledExecutor() .scheduleAtFixedRate( this::send, 0, 3, TimeUnit.SECONDS); Thread.sleep(12*1000); }
@RabbitListener( bindings = @QueueBinding( exchange = @Exchange(value = RabbitMQConstant.DEFAULT_EXCHANGE, type = ExchangeTypes.TOPIC, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), value = @Queue(value = RabbitMQConstant.DEFAULT_QUEUE, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), key = DKEY ), containerFactory = "container", concurrency = "5-10" ) public void process(@Headers Map<String, Object> headers, @Payload ExampleEvent msg) { log.info("basic consumer receive message:{headers = [" + headers + "], msg = [" + msg + "]}"); }
BasicConsumer { @RabbitListener( bindings = @QueueBinding( exchange = @Exchange(value = RabbitMQConstant.DEFAULT_EXCHANGE, type = ExchangeTypes.TOPIC, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), value = @Queue(value = RabbitMQConstant.DEFAULT_QUEUE, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), key = DKEY ), containerFactory = "container", concurrency = "5-10" ) public void process(@Headers Map<String, Object> headers, @Payload ExampleEvent msg) { log.info("basic consumer receive message:{headers = [" + headers + "], msg = [" + msg + "]}"); } }
BasicConsumer { @RabbitListener( bindings = @QueueBinding( exchange = @Exchange(value = RabbitMQConstant.DEFAULT_EXCHANGE, type = ExchangeTypes.TOPIC, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), value = @Queue(value = RabbitMQConstant.DEFAULT_QUEUE, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), key = DKEY ), containerFactory = "container", concurrency = "5-10" ) public void process(@Headers Map<String, Object> headers, @Payload ExampleEvent msg) { log.info("basic consumer receive message:{headers = [" + headers + "], msg = [" + msg + "]}"); }  }
BasicConsumer { @RabbitListener( bindings = @QueueBinding( exchange = @Exchange(value = RabbitMQConstant.DEFAULT_EXCHANGE, type = ExchangeTypes.TOPIC, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), value = @Queue(value = RabbitMQConstant.DEFAULT_QUEUE, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), key = DKEY ), containerFactory = "container", concurrency = "5-10" ) public void process(@Headers Map<String, Object> headers, @Payload ExampleEvent msg) { log.info("basic consumer receive message:{headers = [" + headers + "], msg = [" + msg + "]}"); }  @RabbitListener( bindings = @QueueBinding( exchange = @Exchange(value = RabbitMQConstant.DEFAULT_EXCHANGE, type = ExchangeTypes.TOPIC, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), value = @Queue(value = RabbitMQConstant.DEFAULT_QUEUE, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), key = DKEY ), containerFactory = "container", //指定消费者的线程数量,一个线程会打开一个Channel，一个队列上的消息只会被消费一次（不考虑消息重新入队列的情况） concurrency = "5-10" ) void process(@Headers Map<String, Object> headers, @Payload ExampleEvent msg); }
BasicConsumer { @RabbitListener( bindings = @QueueBinding( exchange = @Exchange(value = RabbitMQConstant.DEFAULT_EXCHANGE, type = ExchangeTypes.TOPIC, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), value = @Queue(value = RabbitMQConstant.DEFAULT_QUEUE, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), key = DKEY ), containerFactory = "container", concurrency = "5-10" ) public void process(@Headers Map<String, Object> headers, @Payload ExampleEvent msg) { log.info("basic consumer receive message:{headers = [" + headers + "], msg = [" + msg + "]}"); }  @RabbitListener( bindings = @QueueBinding( exchange = @Exchange(value = RabbitMQConstant.DEFAULT_EXCHANGE, type = ExchangeTypes.TOPIC, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), value = @Queue(value = RabbitMQConstant.DEFAULT_QUEUE, durable = RabbitMQConstant.FALSE_CONSTANT, autoDelete = RabbitMQConstant.TURE_CONSTANT), key = DKEY ), containerFactory = "container", //指定消费者的线程数量,一个线程会打开一个Channel，一个队列上的消息只会被消费一次（不考虑消息重新入队列的情况） concurrency = "5-10" ) void process(@Headers Map<String, Object> headers, @Payload ExampleEvent msg);  }