写了一个小demo, 用于发送和接收消息,下面创建一个工作队列,向多个消费者分发耗时的任务
图是这样子的
代码学习
生产者
package com.tgb.kwy.workqueues;
import com. rabbitmq .client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Connection factory ;
import com.rabbitmq.client. message Properties;
/**
* Description
*生产
* @author kongwy 15732621629@163.com
* @version 1.0
* @date 2018-07-08-20 -58
*/
public class NewTask {
private static final String TASK_QUEUE_NAME=”task_queue”;
public static void main(String[] args) throws Exception{
ConnectionFactory factory=new ConnectionFactory();
factory.setHost(“192.168.159.132”);/*设置rabbitmq所在主机 ip 或主机名*/
/*指定用户名和密码*/
factory.setUsername(“admin”);
factory.setPassword(“admin”);
Connection connection=factory.newConnection();
Channel channel=connection.createChannel();
channel.queueDeclare(TASK_QUEUE_NAME,true,false,false,null);
String message = getMessage(args);
channel.basicPublish(“”, TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(“UTF-8”));
System.out.println(” [x] Sent ‘” + message + “‘”);
channel. close ();
connection.close();
}
private static String getMessage(String[] strings){
if(strings.length<1){
return “Hello World!”;
}
return joinStrings(strings,” “);
}
private static String joinStrings(String[] strings,String delimiter){
int length=strings.length;
if(length==0){
return “”;
}
StringBuilder words=new StringBuilder(strings[0]);
for (int i = 0; i < length; i++) {
words. append (delimiter).append(strings[i]);
}
return words.toString();
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
消费者代码
package com.tgb.kwy.workqueues;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
* Description
*
* @author kongwy 15732621629@163.com
* @version 1.0
* @date 2018-07-08-21 -03
*/
public class Worker {
private static final String TASK_QUEUE_NAME=”task_queue”;
public static void main(String[] args) throws Exception{
ConnectionFactory factory=new ConnectionFactory();
factory.setHost(“192.168.159.132”);/*设置rabbitmq所在主机ip或主机名*/
/*指定用户名和密码*/
factory.setUsername(“admin”);
factory.setPassword(“admin”);
final Connection connection=factory.newConnection();
final Channel channel=connection.createChannel();
//声明队列,主要为了防止消息接受者先运行此程序,队列还不存在时创建队列
channel.queueDeclare(TASK_QUEUE_NAME,true,false,false,null);
System.out.println(” [*] Waiting for messages. To exit press CTRL+C”);
//channel.basicQos(1); // accept only one unack-ed message at a time (see below)
int prefetchCount=1;
channel.basicQos(prefetchCount);
final Consumer consumer=new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumeTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{
String message=new String(body,”UTF-8″);
System.out.println(” [x] Received ‘” + message + “‘”);
try{
doWork(message);
}finally {
System.out.println(“[x] Done”);
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
boolean autoAck = true; // acknowledgment is covered below
channel.basicConsume(TASK_QUEUE_NAME,autoAck,consumer);
}
private static void doWork(String task){
for(char ch: task.toCharArray()){
if(ch == ‘.’){
try {
Thread.sleep(1000);
}catch (InterruptedException _ignored){
Thread.currentThread().interrupt();
}
}
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
运行结果
任务分发
轮询分发
rabbitmq将逐个发送消息到序列中的写一个消费者,平均每个消费者获得的数量是相同的
Fair dispatch
如果这里有两个消费者的话,可能一个消费者非常的忙碌,但是另一个消费者,几乎一直闲着.但是rabbitmq,还是会均匀地发送消息. 因为当消息进入队列后, rabbitmq就会分配消息,不会看消费者未确认消息的数量.所以为了解决这个问题,就使用了basicQos(prefetchCount=1)方法,来限制rabbitmq只发不超过1条的消息给同一个消费者.当消费者处理完毕后,得到反馈,再发送下一次
注:代码来自官网.官网对于概念性的内容,讲解的还是很清楚的
