RabbitMQ简单应用の轮训分发
2024-10-14 04:59:28
MQ连接工厂还是之前的那个Connection
package com.mmr.rabbitmq.util; import java.io.IOException; import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory; public class ConnectionUtils {
/**
* @desc 获取Mq 的链接
* @author zp
* @throws IOException
* @date 2018-7-19
*/
public static Connection getConnection() throws IOException {
// 1.定义一个链接工厂
ConnectionFactory factroy = new ConnectionFactory(); // 2.设置服务地址
factroy.setHost("127.0.0.1"); // 3.设置端口号
factroy.setPort(5672); // 4.vhost 设置数据库
factroy.setVirtualHost("vhtest"); // 5.设置用户名
factroy.setUsername("jerry"); // 6. 设置密码
factroy.setPassword("123456"); // 7.返回链接
return factroy.newConnection();
}
}
消息生产者类的定义Send
package com.mmr.rabbitmq.work; import java.io.IOException; import com.mmr.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection; public class Send { /*
* |--C1
* P-------|--C2
* |--C3
*
* */
private static final String QUEUE_NAME="test_work_queue";
public static void main(String[] args) throws IOException, InterruptedException{
// 获取链接
Connection connection = ConnectionUtils.getConnection(); // 获取通道
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null); for (int i = 0; i < 50; i++) {
String msg = "hello "+i;
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
System.out.println("send msg 的第"+i+"条");
Thread.sleep(i*20);
}
channel.close();
connection.close();
}
}
消息消费者累的定义 Recv1 Recv2
package com.mmr.rabbitmq.work; import java.io.IOException; import com.mmr.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties; public class Recv1 {
private static final String QUEUE_NAME="test_work_queue";
public static void main(String[] args) throws IOException{
// 获取链接
Connection connection = ConnectionUtils.getConnection(); //获取频道 Channel channel = connection.createChannel(); // 声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 定义一个消费者
Consumer consumer = new DefaultConsumer(channel){
// 一旦有消息 就会触发这个方法 消息到达
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
BasicProperties properties, byte[] body) throws IOException {
// TODO Auto-generated method stub
// 拿消息
String msg = new String(body,"utf-8"); //搭出来
System.out.println("[1]Recv msg:"+msg);
try {
Thread.sleep(2000);
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
}finally{
System.out.println("[1] done");
}
}
};
boolean autoAck = true;
channel.basicConsume(QUEUE_NAME, autoAck,consumer); }
}
package com.mmr.rabbitmq.work; import java.io.IOException; import com.mmr.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties; public class Recv2 {
private static final String QUEUE_NAME="test_work_queue";
public static void main(String[] args) throws IOException{
// 获取链接
Connection connection = ConnectionUtils.getConnection(); //获取频道 Channel channel = connection.createChannel(); // 声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 定义一个消费者
Consumer consumer = new DefaultConsumer(channel){
// 一旦有消息 就会触发这个方法 消息到达
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
BasicProperties properties, byte[] body) throws IOException {
// TODO Auto-generated method stub
// 拿消息
String msg = new String(body,"utf-8"); //搭出来
System.out.println("[2]Recv msg:"+msg);
try {
Thread.sleep(1000);
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
}finally{
System.out.println("[2] done");
}
}
};
boolean autoAck = true;
channel.basicConsume(QUEUE_NAME, autoAck,consumer); }
}
1.首先我们运行Recv1 Recv2 对消息进行监听
2.其次我们运行Send,开始生产消息。
3.最后得到的结果是:消费者1(都是偶数)和消费者2(都是奇数)处理消息是一样的
为什么会出现这种现象呢?
----这种方式叫做轮训分发(round-robin)结果就是不管谁忙谁闲,都不会多给一个消息,任务就是你一个我一个。
最新文章
- Android SharedPreferences存储
- jQuery插件之Cookie插件使用方法~
- zoj 3795 Grouping tarjan缩点 + DGA上的最长路
- Gaussian分布下Hinge损失的期望
- 专访Linux嵌入式开发韦东山操作系统图书作者--转
- 【JavaScript回顾】继承
- Structured Streaming Programming Guide
- JSTL与EL之间的千丝万缕
- C# 内存泄露
- java学习笔记day06---匿名内部类
- delphi idhttp 实战用法(TIdhttpEx)
- 十天学会php第五天
- jdbc_servlet基础增删改分页2(userinfo表的)
- WIFI机器人网
- 【JAVAWEB学习笔记】21_多条件查询、attr和prop的区别和分页的实现
- [js高手之路] 跟GhostWu一起封装一个字符串工具库-扩展字符串位置方法(4)
- UWP Popup 弹出
- 学习python的第一天
- IOS MVC与MVVM的区别
- jsoup对 HTML 文档的解析和操作