欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

RabbitMQ消息队列+spring监听mq服务器多个ip,接收消费mq消息(二)

程序员文章站 2024-03-24 15:57:34
...

前文用了注解方式实现监听多个ip,本文用消费端的类实现ServletContextListener监听器来实现项目启动时开启监听多个ip。大致的代码雷同。
环境和框架:和注解方式完全一样。ssm+maven3.3.9+jdk1.7

1 由于是实现监听器,没有注解,所以并不需要spring的扫包范围限制。我特地把这个监听类放到扫包范围以外来测试。项目结构如下:

RabbitMQ消息队列+spring监听mq服务器多个ip,接收消费mq消息(二)

2 pom.xml中引入rabbitmq的依赖

<dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>1.3.5.RELEASE</version>
        </dependency>

3.com.zhanglf.RabbitMqListenerIpsByImplementsServletContextListener 实现ServletContextListener监听类接口。并在public void contextInitialized()方法中引入监听mq的方法。

package com.zhanglf;

import java.io.IOException;

import javax.servlet.ServletContext;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;

import org.springframework.web.context.WebApplicationContext;
import org.springframework.web.context.support.WebApplicationContextUtils;

import com.rabbitmq.client.Address;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.zlf.bo.StaffBo;
import com.zlf.service.IStaffService;

public class RabbitMqListenerIpsByImplementsServletContextListener implements ServletContextListener {

    private Address[] addre;
    private String vhost;
    private String user;
    private String pwd;
    private String queueName;

    public RabbitMqListenerIpsByImplementsServletContextListener(){
        this.addre = new Address[] { new Address("10.100.82.121", 5672),
                new Address("10.100.82.122", 5672),
                new Address("10.100.82.123", 5672) };
        this.vhost = "gf-iih";
        this.user = "admin";
        this.pwd = "admin";
        this.queueName = "tkq.queue";
    }


    @Override
    public void contextInitialized(ServletContextEvent sce) {
        WebApplicationContext applicationContext = WebApplicationContextUtils.getWebApplicationContext(sce.getServletContext());
        IStaffService staffService = applicationContext.getBean(IStaffService.class);   
        RabbitMqListenerIpsByImplementsServletContextListener mqListener = new  RabbitMqListenerIpsByImplementsServletContextListener();
        mqListener.StartQueueListener(new ShutdownListener() {
            @Override
            public void shutdownCompleted(ShutdownSignalException arg0) {
                System.out.println("Connection Shutdown!");
            }
        }, staffService);
    }

    @Override
    public void contextDestroyed(ServletContextEvent sce) {

    }

    private Connection getConnection(ShutdownListener listener) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername(user);
        factory.setPassword(pwd);
        factory.setVirtualHost(vhost);
        factory.setAutomaticRecoveryEnabled(true);

        Connection conn = null;
        try {
            conn = factory.newConnection(addre);
            conn.addShutdownListener(listener);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return conn;
    }

    public void StartQueueListener(ShutdownListener listener,final IStaffService staffService) {
        Connection conn = getConnection(listener);
        if (conn == null) {
            System.out.println("Failed to Create Connection!");
            return;
        }
        try {
            final Channel channel = conn.createChannel();
            // 设置ACK为手动模式,不在自动ACK
            boolean autoAck = false;
            /**channel.basicConsume各个参数的解释
             * String queueName:队列名
             * boolean autoAck: 服务器是否要手动应答/确认,true-不需要。false-需要。所以这里我们要在处理完业务逻辑后,消费掉mq后发送ack。
             * String consumerTag:用于建立上下文的客户端标签。每个标签都代表一个独立的订阅。同一个channel的不同consumer使用不同的标签。
             * Consumer callback: 消费端接口,实现Consumer的最方便方法是继承DefualtConsumer,并将其作为参数传给basicConsumer方法。答
             */
            channel.basicConsume(queueName, autoAck, "zhanglfConsumerTag",
                    new DefaultConsumer(channel) {
                        @Override
                        public void handleDelivery(
                                String consumerTag,
                                Envelope envelope, 
                                BasicProperties properties,
                                byte[] body)
                        throws IOException {
                            long deliveryTag = envelope.getDeliveryTag();
                            String out = new String(body, "UTF-8");
                            StaffBo staffBo = staffService.selectByPrimaryKey("s01");
                            System.out.println("------------------"+staffBo.getName()+"------------");
                            if ("业务处理结果".equals("业务处理结果")) {
                                // 通知mq服务器移除此条mq。设置ack为每条mq都ack,不是批量ack。
                                channel.basicAck(deliveryTag, false);
                            } else {
                                // 如果业务处理异常,通知服务器回收此条mq。
                                channel.basicNack(deliveryTag, false, true);
                            }
                        }
                    });

        } catch (IOException e) {
            e.printStackTrace();
        }

    }

}

4.在web.xml中引入这个自定义监听类。

<!--自定义监听器 -->
 <listener>  
     <listener-class>com.zhanglf.RabbitMqListenerIpsByImplementsServletContextListener
     </listener-class>  
</listener> 

这样就完成了代码开发,注意点就是要使用WebApplicationContextUtils来获取WebApplicationContext进而可以实例化别的层的类。这里必须这样做因为监听器发生在spring容器初始化之前。当进来启动监听时,要调用的staffService.selectByPrimaryKey("s01"); 还没有实例化,就会报空指针。

WebApplicationContextUtils讲解

WebApplicationContextUtils是一个抽象类,其提供了一个很便利的方法来获取spring应用的上下文即WebApplicationContext。其中的静态方法getWebApplicationContext(ServletContext sc),提供一个ServletContext 类型参数即可。其原理十分简单,在spring容器初始化的方法org.springframework.web.context.ContextLoader.initWebApplicationContext(ServletContext)中通过servletContext.setAttribute(WebApplicationContext.ROOT_WEB_APPLICATION_CONTEXT_ATTRIBUTE, this.context);已经将WebApplicationContext的实例放入ServletContext 中了。然后在工具类的org.springframework.web.context.support.WebApplicationContextUtils.getWebApplicationContext(ServletContext)中就可以通过传入的ServletContext参数获取到WebApplicationContext实例了。

相关标签: rabbitmq监听