java 连接 rabbitMq,并处理数据的例子

青旅半醒 2023-06-30 02:20 3阅读 0赞

从我得另外一篇文章中摘抄出来: [https://mp.csdn.net/postedit/103457479][https_mp.csdn.net_postedit_103457479]

rabbitMq jar版本: amqp-client-5.7.3.jar

## 例子: ##

说明一个rabbitmq 的例子,由于涉及隐私,某些数据已经脱敏,仅供展示

### 连接工厂,创建连接 ###

package mq.util;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * rabbitmq 连接工厂
     */
    public class AmqpConnectFactory {
    
    
        public static String username ="";
        public static String password ="";
        public static String inPlayPackageId = "";
    
        public static String inPlayHost = "";
        
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ClientUitl.startInPlayMq(); // 启动mq
            inPlay();//连接mq 
    
        }
    
    
        /**
         *
         * @throws IOException
         * @throws TimeoutException
         */
        public static void inPlay() throws IOException, TimeoutException {
    //        createConnection(inPlayHost,inPlayPackageId,"C:\\Users\\admin\\Desktop\\newRecord\\");
            ConnectionFactory connectioinFactory = getConnectioinFactory(inPlayHost);
            Connection connection = getConnection(connectioinFactory);
    
            Channel channel =  getChannel(connection,inPlayPackageId,"C:\\Users\\admin\\Desktop\\newRecord\\"); // 开启一个通道并设置 相关的消费者等信息
        }
    
    
    
    
        /**
         * 得到连接工厂
         * @param host
         * @return
         * @throws IOException
         * @throws TimeoutException
         */
        private static ConnectionFactory getConnectioinFactory(String host) throws IOException, TimeoutException {
            ConnectionFactory connectionFactory = new ConnectionFactory();
    
            connectionFactory.setHost(host);
            connectionFactory.setPort(5672);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setAutomaticRecoveryEnabled(true);
            connectionFactory.setVirtualHost("Customers");
            connectionFactory.setRequestedHeartbeat(580);
            connectionFactory.setNetworkRecoveryInterval(1000);// 设置连接间隔
            // 或者使用 url 进行连接
            // factory.setUri（“ amqp：// userName：password @ hostName：portNumber / virtualHost”）;
    //        connectionFactory.setUri("amqp://"+username+":"+password+"@"+ inPlayHost +":" + port +"/" + virtualHost);
    
    //        final Connection conn = connectionFactory.newConnection(); // 新建一个连接
            return connectionFactory;
        }
    
    
        /**
         * 得到一个新的连接
         * @param factory
         * @return
         * @throws IOException
         * @throws TimeoutException
         */
        public static Connection getConnection(ConnectionFactory factory) throws IOException, TimeoutException {
            Connection connection = factory.newConnection();
            return connection;
        }
    
    
        /**
         * 得到一个新的通道
         * @param conn
         * @param packageId
         * @return
         * @throws IOException
         */
        public static Channel getChannel(Connection conn,String packageId,String path) throws IOException {
            final Channel channel = conn.createChannel(); // 打开一个通道(同文档中的模型)
    //        model.BasicQos(prefetchSize: 0, prefetchCount: 1000, global: false);
            channel.basicQos(0, 1000, false);// 配置服务质量
            // 订阅主题, 设置消息自动确认,配置消费者
            channel.basicConsume("_" + packageId +"_", true, new TestConsumer(channel,conn,packageId,path));
            return channel;
        }
    
    
    
    
    
    
    
    
    
    
    
    
    
        @Override
        protected void finalize() throws Throwable {
            super.finalize();
        }
    }

### 消费者配置 ###

这里其实可以用 jar 包自带的DefaultConsumer, 但由于我这边处理比较复杂,因此自己实现了一个

说明: 该例实现了 Consumer 接口的handleConsumeOk() 方法,在消费者配置成功了,(调用定时任务)就调用上文代码中的线程池去队列中取消息, 具体消息处理方法handleDelivery () 方法不再处理具体事务,仅仅将消息放入异步队列中

package mq.util;
    
    import com.rabbitmq.client.*;
    import com.ws.com.util.TimerUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeUnit;
    
    /**
     *   mq 消费者
     *
     */
    public class TestConsumer implements Consumer { //extends DefaultConsumer {
    
    
        private Channel channel;
        private Connection connection;
        private volatile String consumerTag;
    
    
        private String packageId;
        private String path;
    
    
        /**
         * Constructs a new instance and records its association to the passed-in channel.
         *
         * @param channel the channel to which this consumer is attached
         */
        public TestConsumer(Channel channel,Connection connection,String packageId,String path) {
    //        super(channel);
            this.channel = channel;
            this.connection = connection;
            this.packageId = packageId;
            this.path = path;
        }
    
        /**
         * 接收到此使用者的basic.deliver时调用
         * @param consumerTag
         * @param envelope
         * @param properties
         * @param body
         * @throws IOException
         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    //        super.handleDelivery(consumerTag, envelope, properties, body);
            String routingKey = envelope.getRoutingKey();
            String contentType = properties.getContentType();
            long deliveryTag = envelope.getDeliveryTag();
            // 处理消息,将消息放入队列中
            CacheUtil.queue.offer(new String(body));
    //        handleMessage(body,path);
    
            // 由于前面设置通道时已经设置了消息消费后自动确认,所以这里不用再确认消息,否则 双重确认会导致通道异常关闭
    //            channel.basicAck(deliveryTag, true);
    //
        }
    
    //
        /**
         * ：当通道或基础连接被关闭时调用。
         * 通道关闭的问题
         * @param consumerTag
         * @param sig
         */
        @Override
        public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
            System.out.println("通道意外关闭,重新设置通道");
            System.out.println("关闭原因"+sig.getMessage());
    //        this.chan
            try {
                Channel newChannel = connection.createChannel();
                channel = newChannel;// 将绑定的通道设置为新通道
                channel.basicQos(0, 1000, false);// 配置服务质量
                // 配置消费者
                channel.basicConsume("_" + packageId +"_", true, this);
            } catch (IOException e) {
                e.printStackTrace();
            }
    
    
        }
    
        /**
         * 当消费者由于除调用Channel.basicCancel之外的其他原因被取消时调用。
         * @param consumerTag
         * @throws IOException
         */
        @Override
        public void handleCancel(String consumerTag) throws IOException {
    //        super.handleCancel(consumerTag);
            this.consumerTag = consumerTag;
        }
    
        /**
         * 当消费者被Channel.basicCancel的调用取消时调用。
         * @param consumerTag
         */
        @Override
        public void handleCancelOk(String consumerTag) {
    //        super.handleCancelOk(consumerTag);
        }
    
        /**
         * 当使用者通过对任何通道的调用注册时调用 Channel.basicConsume方法。
         * @param consumerTag
         */
        @Override
        public void handleConsumeOk(String consumerTag) {
    //        super.handleConsumeOk(consumerTag);
            System.out.println("注册通过成功");
            // 开启线程池读取消息,并保存, 每 500毫秒调用1次,拿取一个消息
            TimerUtil.setTimer(new SaveRunner(path),50,TimeUnit.MICROSECONDS);
    
        }
    
    
        /**
         *  当 basic.recover. 收到 回复 basic.recover-ok  时调用
         * @param consumerTag
         */
        @Override
        public void handleRecoverOk(String consumerTag) {
            System.out.println("接收消息成功");
    //        super.handleRecoverOk(consumerTag);
        }
    
    
        // ==============================================================
    
    
        public Channel getChannel() {
            return channel;
        }
    
        public void setChannel(Channel channel) {
            this.channel = channel;
        }
    
        public Connection getConnection() {
            return connection;
        }
    
        public void setConnection(Connection connection) {
            this.connection = connection;
        }
    
        public String getConsumerTag() {
            return consumerTag;
        }
    
    
        // ================================================= 具体处理消息的逻辑 ==================================================================================================================
    
    
    
    
    }

其他类,同上篇文章([https://mp.csdn.net/postedit/103457479][https_mp.csdn.net_postedit_103457479]) TimerUtil, CacheUtil, SaveRunner

[https_mp.csdn.net_postedit_103457479]: https://mp.csdn.net/postedit/103457479