spring boot 和 BlockingQueue 监听器

Question

我已经用spring boot实现了jms，我正在使用@JmsListener来监听主题

  @Component
    public class AMQListner {
        BlockingQueue<MessageBO> queue = new ArrayBlockingQueue<>(1024);
        @JmsListener(destination = "${spring.activemq.topic}")
        public void Consume(TextMessage message) {
            try {
                String json = message.getText();
                MessageBO bo = ObjectMapperConfig.getInstance().readValue(json, MessageBO.class);
                queue.add(bo);
            } catch (JMSException e) {
                e.printStackTrace();
            } catch (JsonParseException e) {
                e.printStackTrace();
            } catch (JsonMappingException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

现在我想要一个侦听器来侦听该阻塞队列，如果它具有 value ，则处理。我们可以在 Spring Boot 中使用注释来实现这一点吗？

Answer 1

首先，正确的方法是创建一个处理程序 bean，而不是在接收器类中拥有消息队列的成员。

public interface MessageHandler extends Consumer<MessageBO> {
    public default void handle(MessageBO msg) { accept(msg); }
}

@Component
public class AMQListener {
    @Resource("multiplexer")
    MessageHandler handler;

    @JmsListener(destination = "${spring.activemq.topic}")
    public void Consume(TextMessage message) {
        try {
            String json = message.getText();
            MessageBO bo = ObjectMapperConfig.getInstance().readValue(json, MessageBO.class);
            handler.handle(bo);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

然后，您将在处理程序 bean 中拥有队列

@Component("multiplexer")
public class MessageMultiplexer implements MessageHandler {
    @Autowired
    MessageHandler actualConsumer;

    ExecutorService executor = Executors.newFixedThreadPool(4);
    public void accept(MessageBO msg) {
        executor.submit(msg -> actualConsumer.handle(msg));
    }
}

在这种情况下，Executor 几乎就是队列。

警告：您没有这种方式的 1024 限制。您可以通过使用ThreadPoolExecutor 构造函数并传递给它一个有限的队列来做到这一点。