_amq_group_id存在于消息中，但@jmslistener中的JMSXGroupID为空

Question

消息组中的消息共享相同的组id，即它们具有相同的组标识符属性（JMS为JMSXGroupID，Apache ActiveMQ Artemis Core API为_AMQ_GROUP_ID)。

当我浏览值为product=paper的代理中的消息时，我可以看到为什么最初通过jmsxgroupid设置的属性变成了_amq_group_id。但是，在我的@jmslistener注释方法中，我可以看到_amq_group_id属性丢失，并且jmsxgroupid在消息的headershashmap中显示为null。

@JmsListener(destination = "${artemis.destination}", subscription = "${artemis.subscriptionName}",
            containerFactory = "containerFactory", concurrency = "15-15")
public void consumeMessage(Message<StatefulSpineEvent<?>> eventMessage)

所以

null

@Component
public class GroupIdMessageMapper extends SimpleJmsHeaderMapper {

    @Override
    public MessageHeaders toHeaders(Message jmsMessage) {

        MessageHeaders messageHeaders = super.toHeaders(jmsMessage);

        Map<String, Object> messageHeadersMap = new HashMap<>(messageHeaders);

        try {
            messageHeadersMap.put("JMSXGroupID", jmsMessage.getStringProperty("_AMQ_GROUP_ID"));
        } catch (JMSException e) {
            e.printStackTrace();
        }

        // can see while debugging that this returns the correct headers
        return new MessageHeaders(messageHeadersMap);
    }
}

@Component
public class CustomSpringJmsListener {

    protected final Logger LOG = LoggerFactory.getLogger(getClass());

    @JmsListener(destination = "local-queue", subscription = "groupid-example",
            containerFactory = "myContainerFactory", concurrency = "15-15")
    public void receive(Message message) throws JMSException {
        LOG.info("Received message: " + message);
    }
}

@SpringBootApplication
@EnableJms
public class GroupidApplication implements CommandLineRunner {

    private static Logger LOG = LoggerFactory
            .getLogger(GroupidApplication.class);

    @Autowired
    private JmsTemplate jmsTemplate;

    @Autowired MessageConverter messageConverter;

    public static void main(String[] args) {
        LOG.info("STARTING THE APPLICATION");
        SpringApplication.run(GroupidApplication.class, args);

        LOG.info("APPLICATION FINISHED");
    }

    @Override
    public void run(String... args) {
        LOG.info("EXECUTING : command line runner");

        jmsTemplate.setPubSubDomain(true);

        createAndSendObjectMessage("Message1");
        createAndSendTextMessage("Message2");
        createAndSendTextMessage("Message3");
        createAndSendTextMessage("Message4");
        createAndSendTextMessage("Message5");
        createAndSendTextMessage("Message6");
    }

    private void createAndSendTextMessage(String messageBody) {
        jmsTemplate.send("local-queue", session -> {
            Message message = session.createTextMessage(messageBody);

            message.setStringProperty("JMSXGroupID", "product=paper");

            return message;
        });
    }

    // BEANS

    @Bean
    public JmsListenerContainerFactory<?> myContainerFactory(ConnectionFactory connectionFactory,
            DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        // This provides all boot's default to this factory, including the message converter
        configurer.configure(factory, connectionFactory);
        // You could still override some of Boot's default if necessary.
        factory.setSubscriptionDurable(true);
        factory.setSubscriptionShared(true);
        factory.setMessageConverter(messagingMessageConverter());

        return factory;
    }

    @Bean
    public MessagingMessageConverter messagingMessageConverter() {
        return new MessagingMessageConverter(messageConverter, new GroupIdMessageMapper());
    }
}

SimpleJmsHeaderMapper调用位置的堆栈跟踪：

ToHeaders：130,SimpleJmsHeaderMapper（org.springframework.jms.support）ToHeaders：57,SimpleJmsHeaderMapper（org.springframework.jms.support）ExtractHeaders：148,MessagingMessageConverter（org.springframework.jms.support.converter）Access$100:466,AbstractAdaptableMessageConverterAdapter（eargumentinternal:68，HeaderMethodArgumentResolver(org.springframework.messaging.handler.annotation.support）resolveargument:100，AbstractNamedValueMethodArgumentResolver(org.springframework.messaging.handler.annotation.support）resolveargument:117，HandlerMethodArgumentResolver(org.springframework.messaging.handler.annotation.support）framework.messaging.handler.invocation）InvokeHandler:114,MessagingMessageListenerAdapter（org.springframework.jms.listener.adapter）onMessage:77,MessagingMessageListenerAdapter（org.springframework.jms.listener.adapter）doInvokeListener:736,AbstractMessageListenerContainer（org.springframework.jms.listener）InvokeListener:696,IstenerContainer（org.springframework.jms.listener）ReceiveandExecute:257,AbstractPollingMessageListenerContainer（org.springframework.jms.listener）InvokeListener:1190,DefaultMessageListenerContainer$AsyncMessageListenerInvoker（org.springframework.jms.listener）ExecuteOnGoingLoop:1180,DefaultMessageListenerContainer.springframework.jms.listener）运行：748,Thread（java.lang）

Answer 1

尝试子类化SimpleJMSheaderMapper并重写ToHeaders()。调用super.toHeaders()，根据结果创建一个新的映射<>；将()所需的任何其他标头放入映射中，并从映射返回一个新的MessageHeaders。

将自定义映射器传递到新的MessagingMessageConverter并将其传递到容器工厂。

如果使用Spring Boot，只需将转换器添加为@bean，Boot将自动将其连接到工厂中。

@SpringBootApplication
public class So58399905Application {

    public static void main(String[] args) {
        SpringApplication.run(So58399905Application.class, args);
    }

    @JmsListener(destination = "foo")
    public void listen(String in, MessageHeaders headers) {
        System.out.println(in + headers);
    }

    @Bean
    public ApplicationRunner runner(JmsTemplate template) {
        return args -> template.convertAndSend("foo", "bar", msg -> {
            msg.setStringProperty("JMSXGroupID", "product=x");
            return msg;
        });
    }

}

bar{jms_redelivered=false, JMSXGroupID=product=x, jms_deliveryMode=2, JMSXDeliveryCount=1,  ...