RabbitMQ系列之二：work queue

 1 package com.example.workqueue;
 2 
 3 import java.io.IOException;
 4 
 5 import com.rabbitmq.client.Channel;
 6 import com.rabbitmq.client.Connection;
 7 import com.rabbitmq.client.ConnectionFactory;
 8 import com.rabbitmq.client.MessageProperties;
 9 
10 public class Send {
11 
12     public static void main(String[] args) throws IOException {
13         
14         // 队列名称
15         String queueName = "task_queue";
16         
17         ConnectionFactory factory = new ConnectionFactory();
18         
19         //远程服务器ip，如果在本地测试可以改成localhost
20         factory.setHost("121.40.151.120");
21         
22         //不是在本地测试，用户名和密码必填
23         factory.setUsername("rabbitmqname");
24         factory.setPassword("rabbitmqpwd");
25         
26         Connection conn = factory.newConnection();
27         Channel channel = conn.createChannel();
28         
29         boolean durable = true;  
30         
31         /** 
32          * 参数说明：
33          * queue:队列名称
34          * durable：队列数据是否可以持久化，true：是，false：否。也就是服务重启后队列数据是否依然存在
35          * exclusive：是否为某一个队列的专用连接
36          * autoDelete：当队列不再被使用也就是没有消费者的时候是否自动删除
37          * arguments：其它参数，比如队列存活时间
38         */
39         channel.queueDeclare(queueName, durable, false, false, null);
40         
41         String[] strs = new String[] { "First message." }; 
42         String message = getMessage(strs);  
43         
44         /** 
45          * 参数说明：
46          * exchange:默认的exchange就是"",是direct类型的，
47          *             任何发往到默认exchange的消息都会被路由到routingKey的名字对应的队列上，如果没有对应的队列，则消息会被丢弃。
48          * routingKey：指定接收消息的队列
49          * props：其它属性，比如消息路由头信息,持久化信息
50          * body：消息内容
51         */  
52         channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());  
53         
54         System.out.println("[" + message + "]");  
55           
56         // 最后，我们关闭channel和连接，释放资源。  
57         channel.close();  
58         conn.close(); 
59     }
60     
61     private static String getMessage(String[] strings) {  
62         if (strings.length < 1) {  
63             return "Hello World!";  
64         }  
65         return joinStrings(strings, " ");  
66     }  
67   
68     private static String joinStrings(String[] strings, String delimiter) {  
69         int length = strings.length;  
70         if (length == 0) {  
71             return "";  
72         }  
73         StringBuilder words = new StringBuilder(strings[0]);  
74         for (int i = 1; i < length; i++) {  
75             words.append(delimiter).append(strings[i]);  
76         }  
77         return words.toString();  
78     } 
79 
80 }

 1 package com.example.workqueue;
 2 
 3 import java.io.IOException;
 4 
 5 import com.rabbitmq.client.Channel;
 6 import com.rabbitmq.client.Connection;
 7 import com.rabbitmq.client.ConnectionFactory;
 8 import com.rabbitmq.client.ConsumerCancelledException;
 9 import com.rabbitmq.client.QueueingConsumer;
10 import com.rabbitmq.client.ShutdownSignalException;
11 
12 public class Recv {
13 
14     public static void main(String[] args) throws IOException, ShutdownSignalException, 
15         ConsumerCancelledException, InterruptedException {
16         
17         // 队列名称
18         String queueName = "task_queue";
19         
20         ConnectionFactory factory = new ConnectionFactory(); 
21         
22         factory.setHost("121.40.151.120");
23         factory.setUsername("rabbitmqname");
24         factory.setPassword("rabbitmqpwd");
25         
26         Connection connection = factory.newConnection();  
27         Channel channel = connection.createChannel();  
28         
29         // 表示在同一时间不要给一个Rev一个以上的消息（只能是一个），也就是说不要将一个新的消息分发给Rev直到它处理完了并且返回了前一个消息的通知标志（acknowledged）
30         channel.basicQos(1);
31           
32         //与服务端一致
33         channel.queueDeclare(queueName, true, false, false, null);  
34         
35         System.out.println("CRTL+C");  
36           
37         // QueueingConsumer：用来缓存服务端推送给我们的消息。  
38         QueueingConsumer consumer = new QueueingConsumer(channel);  
39         
40         boolean autoAck = false;
41         /** 
42          * 参数说明：
43          * queue:队列名称
44          * autoAck：是否自动应答，true：消息一旦被消费者消费，服务端就知道该消息已经投递，从而从队列中将消息剔除；
45          *                      false：需要在消费端显示调用channel.basicAck()方法通知服务端，如果没用显示调用，消息将进入
46          *                            unacknowledged状态，当前消费者连接断开后该消息变成ready状态重新进入队列。
47          * callback:具体消费者类
48         */
49         channel.basicConsume(queueName, autoAck, consumer);  
50           
51         while (true) {  
52             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
53             String message = new String(delivery.getBody());  
54             System.out.println("[" + message + "]");  
55             doWork(message);
56             System.out.println("r[done]");
57             
58             /**
59              * 显示调用通知服务端该消息已经消费并返回了acknowledged
60              * true:通知所有相同tag的untracked，false：只通知当前一个
61              */
62             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); 
63         } 
64     }
65 
66     private static void doWork(String message) throws InterruptedException {
67         for (char ch : message.toCharArray()) {  
68             if (ch == ‘.‘) {  
69                 Thread.sleep(1000);  
70             }  
71         }
72     }
73 
74 }