账号密码登录
微信安全登录
微信扫描二维码登录

登录后绑定QQ、微信即可实现信息互通

手机验证码登录
找回密码返回
邮箱找回 手机找回
注册账号返回
其他登录方式
分享
  • 收藏
    X
    spring boot rabbitmq集成,只能监听到两条消息后,然后没有了
    37
    0

    目前mq中存放到消息数量有几千条。集成到项目中,每次启动之后,只能够consume掉2条之后,就自动不消耗了。。。。但是单独将consumer拎出来,成为一个纯净到单独到project,又是可以不断监听到mq中的消息,不断消费。原因真是未知。

    @Configuration
    @EnableRabbit
    public class RabbitMQConfiguration{
        @Value("${host}")
        private String host;
    
        @Value("${port}")
        private int port;
    
        @Value("${username}")
        private String username;
    
        @Value("${password}")
        private String password;
    
        @Value("${vhost}")
        private String virtualHost;
    
        @Value("${routing_key}")
        private String routing_key;
    
        @Value("${queue_name}")
        private String queue_name;
    
        @Value("${exchange}")
        private String exchange;
    
        @Value("${queue_durable}")
        private boolean queue_durable;
    
        @Value("${exchange_durable}")
        private boolean exchange_durable;
    
        @Value("${exchange_autoDelete}")
        private boolean exchange_autoDelete;
    
       
    
        @Bean
        public ConnectionFactory connectionFactory() throws Exception {
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setPort(port);
            connectionFactory.setHost(host);
            connectionFactory.setVirtualHost(virtualHost);
            connectionFactory.afterPropertiesSet();
            return connectionFactory;
        }
    
    
        @Bean(name = "springRabbitTemplate")
        @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
        public RabbitTemplate RabbitTemplate() throws Exception {
            RabbitTemplate template = new RabbitTemplate(connectionFactory());
            return template;
        }
    
        @Bean
        public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
            RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
            return rabbitAdmin;
        }
    
        @Bean
        public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
            //SimpleRabbitListenerContainerFactory发现消息中有content_type有text就会默认将其转换成string类型的
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            return factory;
        }
    }
    @EnableRabbit
    @Slf4j
    @SpringBootApplication
    @ComponentScan
    public class Application implements CommandLineRunner {
    
    
        @Value("${host}")
        private String host;
    
        @Value("${port}")
        private int port;
    
        @Value("${username}")
        private String username;
    
        @Value("${password}")
        private String password;
    
        @Value("${vhost}")
        private String virtualHost;
    
        @Value("${routing_key}")
        private String routing_key;
    
        @Value("${queue_name}")
        private String queue_name;
    
        @Value("${exchange}")
        private String exchange;
    
        @Value("${queue_durable}")
        private boolean queue_durable;
    
        @Value("${exchange_durable}")
        private boolean exchange_durable;
    
        @Value("${exchange_autoDelete}")
        private boolean exchange_autoDelete;
    
    
        @Bean
        public Queue queue() {
            return new Queue(queue_name, queue_durable);
        }
    
        @Bean
        public DirectExchange exchange() {
            return new DirectExchange(exchange,exchange_durable,exchange_autoDelete);
        }
    
        @Bean
        public Binding binding(Queue queue, DirectExchange exchange) {
            return BindingBuilder.bind(queue).to(exchange).with(routing_key);
        }
    
    
        @RabbitListener(bindings ={@QueueBinding(value = @org.springframework.amqp.rabbit.annotation.Queue(value = "${queue_name}",durable = "${queue_durable}"),
                exchange =@org.springframework.amqp.rabbit.annotation.Exchange(value = "${exchange}",durable = "${exchange_durable}", type=ExchangeTypes.DIRECT),
                key="${routing_key}")})
        @RabbitHandler
        public void receivePaymentMsg(Message message){
            System.out.println("---!!!!!!!!!!-------"+message.getBody());
            JSONObject object = JSON.parseObject(new String(message.getBody()));
            log.info(object.toJSONString());
            JSONObject payload= object.getJSONObject("payload");
            if(payload.containsKey("id")) {
                String id = payload.get("id").toString();
                System.out.println(id);
            }
        }
    
        public static void main(String[] args) {
            SpringApplication.run(Application.class, args).close();
        }
    
        @Override
        public void run(String... args) {
            System.out.println("consumer start running...");
        }

    现象:

    输出这样的,然后没有了

    ---!!!!!!!!!!-------[B@7ea7b053
    40ab
    ticket consumer start running...
    ---!!!!!!!!!!-------[B@44fccb7a
    40ac

    并且在rabbitmq的管理界面,queue那里,consumer没有任何consumer注册上了,是空的。

    1
    打赏
    收藏
    点击回答
        全部回答
    • 0
    • 龙飞云翔 普通会员 1楼

      在Spring Boot中,可以通过RabbitTemplateRabbitListener来实现RabbitMQ的集成。RabbitTemplate用于发送和接收消息,而RabbitListener用于处理收到的消息。

      当你的消息被发送到RabbitMQ服务器时,RabbitTemplate会创建一个新的MessageListener,并监听这个MessageListener。当收到一个新的Message时,MessageListener会处理这个Message,然后将处理后的消息发送回RabbitMQ服务器。

      但是,你可能会遇到以下情况:

      1. 如果你的MessageListener没有处理新的Message,那么RabbitTemplate就不会再创建新的MessageListener,消息就会被发送回RabbitMQ服务器。

      2. 如果你的MessageListener没有处理新的Message,那么RabbitTemplate就会发送新的Message给RabbitMQ服务器,而不会再创建新的MessageListener。

      为了解决这个问题,你可以添加一些代码来检查MessageListener是否已经创建。如果已经创建,那么就等待新的Message。如果还没有创建,那么就创建一个新的MessageListener。

      以下是一个简单的示例:

      ```java import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.annotation.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;

      @Component public class RabbitListenerExample {

      @Autowired
      private RabbitTemplate rabbitTemplate;
      
      @RabbitListener("your.rabbitmq.queue")
      public void processMessage(String message) {
          // Process the message
      }
      

      } ```

      在这个示例中,我们有一个@RabbitListener注解,它告诉Spring Boot我们想要监听"your.rabbitmq.queue"队列。然后,我们在processMessage方法中处理收到的消息。

    更多回答
    扫一扫访问手机版
    • 回到顶部
    • 回到顶部