Weblogic JMS 集群服务器缺少给消费者的传递消息

如何解决Weblogic JMS 集群服务器缺少给消费者的传递消息

我在我的系统中使用 Weblogic JMS 集群服务器作为消息代理。我的消费者使用 Spring Boot 并通过注解 @JmsListener 注册 JMS 侦听器。在 Weblogic 上使用 Cluster JMS 服务器运行时，我会错过消息。我已经检查过，发布者每次都已经将消息发布到 JMS 集群，但有时我没有收到消费者的消息。

这是我的消费者应用程序的配置：

@JmsListener(destination = "${spring.jms.jndi-name.listener.int.test}",containerFactory = "workerJmsListenerContainerFactory")
    public void processListenerTestMessage(TextMessage message){
        logger.info("Received Int Listener Test Message - msg counter  " + msgCount);
        msgCount++;
    }

这是我的 JMS 配置（JMS 容器工厂和连接工厂...等）：

@Configuration
@EnableJms
public class JMSConfiguration {

    private Logger logger = LoggerFactory.getLogger(JMSConfiguration.class);

    @Value("${spring.jms.int.url}")
    private String jmsIntUrl;
    @Value("${spring.jms.int.username}")
    private String jmsIntUser;
    @Value("${spring.jms.int.password}")
    private String jmsIntPassword;

    @Value("${spring.jms.connectionFactoryName}")
    private String jmsConnFactoryName;

    private Properties getJNDIProperties(){

        final Properties jndiProps = new Properties();
        jndiProps.setProperty(Context.INITIAL_CONTEXT_FACTORY,"weblogic.jndi.WLInitialContextFactory");
        jndiProps.setProperty(Context.PROVIDER_URL,jmsIntUrl);
        logger.info("url " + jmsIntUrl + " username " + jmsIntUser + " pass " + jmsIntPassword);
        if (username != null && !username.isEmpty()) {
            jndiProps.setProperty(Context.SECURITY_PRINCIPAL,jmsIntUser);
        }
        if (password != null && !password.isEmpty()) {
            jndiProps.setProperty(Context.SECURITY_CREDENTIALS,jmsIntPassword);
        }
        return jndiProps;
    }

    @Bean(name = "queueConnectionFactory")
    @Autowired
    public QueueConnectionFactory connectionFactory(@Qualifier("jndiTemplate") JndiTemplate jndiTemplate) throws NamingException {
        // JNDI connection factory name stored in weblogic.
        QueueConnectionFactory connectionFactory = lookupByJndiTemplate(jmsConnFactoryName,QueueConnectionFactory.class,jndiTemplate);
        logger.info("connectionFactory " + connectionFactory);
        return connectionFactory;
    }

    @Bean(name = "cachingConnectionFactory")
    @Autowired
    public CachingConnectionFactory cachingConnectionFactory(@Qualifier("queueConnectionFactory") QueueConnectionFactory queueConnectionFactory) throws NamingException {
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(queueConnectionFactory);

        cachingConnectionFactory.setReconnectOnException(true);
        cachingConnectionFactory.setSessionCacheSize(20);
        return cachingConnectionFactory;
    }

    @Bean(name = "workerJmsListenerContainerFactory")
    @Autowired
public JmsListenerContainerFactory<?> containerFactory(@Qualifier("cachingConnectionFactory") CachingConnectionFactory cachingConnectionFactory,DefaultJmsListenerContainerFactoryConfigurer configurer,@Qualifier("jmsDestinationResolver") JndiDestinationResolver jndiDestinationResolver) throws NamingException {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
//        factory.setConnectionFactory(queueConnectionFactory);
        factory.setConnectionFactory(cachingConnectionFactory);
        factory.setSessionTransacted(true);
        factory.setConcurrency("3-10");
        factory.setDestinationResolver(jndiDestinationResolver);
        factory.setErrorHandler(new ErrorHandler() {
            @Override
            public void handleError(Throwable throwable) {
                logger.warn("jms error handle ");
                logger.error("Error Message : {}",throwable.getMessage());
            }
        });
        configurer.configure(factory,cachingConnectionFactory);
        return factory;
    }

    @Bean(name = "jmsDestinationResolver")
    @Autowired
    public JndiDestinationResolver jmsDestinationResolver(@Qualifier("jndiTemplate") JndiTemplate jndiTemplate){
        JndiDestinationResolver resolver = new JndiDestinationResolver();
        resolver.setJndiTemplate(jndiTemplate);
        resolver.setCache(true);
        return resolver;
    }

    @Bean(name = "jndiTemplate")
    public JndiTemplate jndiTemplate() {
        JndiTemplate jndiTemplate = new JndiTemplate();
        jndiTemplate.setEnvironment(getJNDIProperties());
        logger.info("jndiTemplate " + jndiTemplate);
        return jndiTemplate;
    }

    protected <T> T lookupByJndiTemplate(String jndiName,Class<T> requiredType,JndiTemplate jndiTemplate) {

        try {
            Object located = jndiTemplate.lookup(jndiName);
            logger.info("lookupByJndiTemplate located " + located);
            if (located == null) {
                throw new NameNotFoundException("JNDI object with [" + jndiName + "] not found");
            }
            return (T) located;
        } catch (NamingException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Bean
    public MessageConverter jacksonJmsMessageConverter(){
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setTargetType(MessageType.TEXT);
        converter.setTypeIdPropertyName("_type");
        return converter;
    }
}

我的 JMS 服务器配置或消费者配置缺少什么以使用集群 JMS 上的消息