java-rabbitmq-实例pull模式拉取消息

男娘i 2024-02-18 15:24 62阅读 0赞

java-rabbitmq-实例pull模式拉取消息  
  
  
描述：  
手动拉取指定队列的消息。  
  
  
运行：  
D7\_PullSend.main();  
D7\_PullRecv.main();

package com.example.tutorials;
    
    
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Scanner;
    
    
    /**
     * 使用 channel.basicGet() 方法，拉取指定队列中的内容
     * @create 2017-09-05
     * amqp-client 4.2.0
     **/
    public class D7_PullSend {
        private final static String QUEUE_NAME = "pull_queue";
    
    
        /**
         * 生产者，
         * @param argv
         * @throws Exception
         */
        public static void main(String[] argv) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            //设置登录账号
            factory.setHost(ServerInfo.host);
            factory.setPort(ServerInfo.port);
            factory.setUsername(ServerInfo.uName);
            factory.setPassword(ServerInfo.uPwd);
            //链接服务器
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
    
            Map<String,Object> args = new HashMap<>();
    //        args.put("x-message-ttl",15*1000);//消息过期时间为 15 秒
            //args.put("x-expires",1*60*1000); //队列过期时间为 1 分钟
            //定义一个队列
            boolean duiable=false;//持久化
            boolean exclusive = false;//排他队列
            boolean autoDelete=false;//没有consumer时，队列是否自动删除
            channel.queueDeclare(QUEUE_NAME, duiable, exclusive, autoDelete, args);
    
    
            //发送消息
            System.out.println("输入要发送的消息，退出输入 x ");
            String message ;
            //
            AMQP.BasicProperties prop = new AMQP.BasicProperties
                    .Builder()
                    .expiration("15000") //消息过期时间为 15 秒
                    .build();
            do{
                Scanner scanner = new Scanner(System.in);
                message = scanner.next();
                channel.basicPublish(""
                        , QUEUE_NAME
                        , prop
                        , message.getBytes("UTF-8"));
                System.out.println(" [x] Sent '" + message + "'");
            }while(!"x".equals(message));
            //关闭链接
            channel.close();
            connection.close();
        }
    }

package com.example.tutorials;
    
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.GetResponse;
    
    
    import java.util.Scanner;
    
    
    /**
     * 使用 channel.basicGet() 方法，拉取指定队列中的内容
     * @create 2017-09-05
     * amqp-client 4.2.0
     **/
    public class D7_PullRecv {
        private final static String QUEUE_NAME = "pull_queue";
    
    
        /**
         * 消费者， "Hello World!"
         * @param argv
         * @throws Exception
         */
        public static void main(String[] argv) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            //设置登录账号
            factory.setHost(ServerInfo.host);
            factory.setPort(ServerInfo.port);
            factory.setUsername(ServerInfo.uName);
            factory.setPassword(ServerInfo.uPwd);
            //链接服务器
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            //定义一个队列
            boolean duiable=false;//持久化
            boolean exclusive = false;//排他队列
            boolean autoDelete=false;//没有consumer时，队列是否自动删除
            channel.queueDeclare(QUEUE_NAME, duiable, exclusive, autoDelete, null);
    
    
            //接收消息
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
            boolean autoAck = true; //自动应答
    
    
            System.out.println("输入回车获取消息，退出输入 x ");
            String message ="";
            GetResponse resp ;
            do{
                Scanner scanner = new Scanner(System.in);
                scanner.nextLine();
                resp = channel.basicGet(QUEUE_NAME,autoAck);
                if(resp==null){
                    System.out.println(QUEUE_NAME+" 队列无消息");
                    continue;
                }
                message = new String(resp.getBody(), "UTF-8");
                System.out.println(String.format(" [x] Recv Count %s , msg = %s;"
                    ,resp.getMessageCount()
                    ,message));
            }while(!"x".equals(message));
        }
    }