代码非本人所写,只是将代码稍微做了些改动,在自己的环境中运行起来了,侵删
模型图是这种模式的模型图,并不是代码对应的模型图,仅供参考
ConnectionUtils.java
创建并设置factory的基本属性,并返回新创建的connection
package test.rabbitmq.rabbitmq.mq5s;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ConnectionUtils {
public static Connection getConnection() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
return factory.newConnection();
}
}
模型图
发送端:Send.java
package test.rabbitmq.rabbitmq.mq5s.mq1;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import test.rabbitmq.rabbitmq.mq5s.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**简单队列
* 生产者发送消息
*/
public class Send {
private static final String QUEUE_NAME = "test_simple_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String msg = "hello simple!!!!!!!!!!!";
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
System.out.println("--send msg;" + msg);
channel.close();
connection.close();
}
}
接收端:Receive.java
package test.rabbitmq.rabbitmq.mq5s.mq1;
import com.rabbitmq.client.*;
import test.rabbitmq.rabbitmq.mq5s.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者获取消息
*/
public class Receive {
private static final String QUEUE_NAME = "test_simple_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//获取链接
Connection connection = ConnectionUtils.getConnection();
//创建通道
Channel channel = connection.createChannel();
//队列声明
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
DefaultConsumer consumer = new DefaultConsumer(channel) {
//获取到达的消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
String msg = new String(body, "utf-8");
System.out.println("new api recv:" + msg);
}
};
//监听队列
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
模型图
发送端:Send.java
package test.rabbitmq.rabbitmq.mq5s.mq2;
import com.rabbitmq.client.*;
import test.rabbitmq.rabbitmq.mq5s.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**轮询分发
* \---c1
* p---Queue----\
* \---c2
*/
public class Send {
private static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//获取链接
Connection connection = ConnectionUtils.getConnection();
//获取channel
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
for (int i = 0; i < 100; i++) {
String msg = "hello" + i;
System.out.println("[WQ ] send:" + msg);
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
Thread.sleep(i * 20);
}
channel.close();
connection.close();
}
}
接收端01:Receive1.java
package test.rabbitmq.rabbitmq.mq5s.mq2;
import com.rabbitmq.client.*;
import test.rabbitmq.rabbitmq.mq5s.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receive1 {
public static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//创建链接
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, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, "utf-8");
System.out.println("[1] Recv msg :" + msg);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[1] done");
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
接收端02:Receive2.java
package test.rabbitmq.rabbitmq.mq5s.mq2;
import com.rabbitmq.client.*;
import test.rabbitmq.rabbitmq.mq5s.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receive2 {
public static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//创建链接
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, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, "utf-8");
System.out.println("[2] Recv msg :" + msg);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[2] done");
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
模型图
发送端:Send.java
package test.rabbitmq.rabbitmq.mq5s.mq3;
import com.rabbitmq.client.*;
import test.rabbitmq.rabbitmq.mq5s.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**公平分发
* \---c1
* p---Queue----\
* \---c2
*/
public class Send {
private static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//获取链接
Connection connection = ConnectionUtils.getConnection();
//获取channel
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
/**
* 每个消费者发送消费之前,消息队列不发送下一个消息到消费者,一次只处理一个消息
* 限制发送给同一个消费者不得超过一个消息
*/
int prefetchCount = 1;
channel.basicQos(prefetchCount);
for (int i = 0; i < 50; i++) {
String msg = "hello" + i;
System.out.println("[WQ ] send:" + msg);
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
Thread.sleep(i * 5);
}
channel.close();
connection.close();
}
}
接收端01:Receive1.java
package test.rabbitmq.rabbitmq.mq5s.mq3;
import com.rabbitmq.client.*;
import test.rabbitmq.rabbitmq.mq5s.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receive1 {
public static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//创建链接
Connection connection = ConnectionUtils.getConnection();
//创建频道
final Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//保证一次只发送一个
channel.basicQos(1);
//定义一个消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, "utf-8");
System.out.println("[1] Recv msg :" + msg);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[1] done");
//手动回执
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
接收端02:Receive2.java
package test.rabbitmq.rabbitmq.mq5s.mq3;
import com.rabbitmq.client.*;
import test.rabbitmq.rabbitmq.mq5s.ConnectionUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receive2 {
public static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//创建链接
Connection connection = ConnectionUtils.getConnection();
//创建频道
final Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//保证一次只发送一个
channel.basicQos(1);
//定义一个消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, "utf-8");
System.out.println("[2] Recv msg :" + msg);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[2] done");
//手动回执
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
模型图
发送端:send.java
package test.rabbitmq.rabbitmq.mq5s.mq4;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 发送消息
*/
public class send {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//2. 设置参数
factory.setHost("127.0.0.1");//ip 默认值 localhost
factory.setPort(5672); //端口 默认值 5672
factory.setVirtualHost("/");//虚拟机 默认值/
factory.setUsername("guest");//用户名 默认 guest
factory.setPassword("guest");//密码 默认值 guest
//3. 创建连接 Connection
Connection connection = factory.newConnection();
//4. 创建Channel
Channel channel = connection.createChannel();
/*
exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
参数:
1. exchange:交换机名称
2. type:交换机类型
DIRECT("direct"),:定向,把消息交给符合指定routing key 的队列。
FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。
TOPIC("topic"),通配符的方式
HEADERS("headers");参数匹配
3. durable:是否持久化
4. autoDelete:自动删除
5. internal:内部使用。 一般false
6. arguments:参数
*/
String exchangeName = "test_fanout";
//5. 创建交换机
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
//6. 创建队列
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//7. 绑定队列和交换机
/*
queueBind(String queue, String exchange, String routingKey)
参数:
1. queue:队列名称
2. exchange:交换机名称
3. routingKey:路由键,绑定规则
如果交换机的类型为fanout ,routingKey设置为""
*/
channel.queueBind(queue1Name,exchangeName,"");
channel.queueBind(queue2Name,exchangeName,"");
String body = "日志信息:张三调用了findAll方法...日志级别:info...";
//8. 发送消息
channel.basicPublish(exchangeName,"",null,body.getBytes());
//9. 释放资源
channel.close();
connection.close();
}
}
接收端01:receive1.java
package test.rabbitmq.rabbitmq.mq5s.mq4;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class receive1 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//2. 设置参数
factory.setHost("127.0.0.1");//ip 默认值 localhost
factory.setPort(5672); //端口 默认值 5672
factory.setVirtualHost("/");//虚拟机 默认值/
factory.setUsername("guest");//用户名 默认 guest
factory.setPassword("guest");//密码 默认值 guest
//3. 创建连接 Connection
Connection connection = factory.newConnection();
//4. 创建Channel
Channel channel = connection.createChannel();
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
/*
basicConsume(String queue, boolean autoAck, Consumer callback)
参数:
1. queue:队列名称
2. autoAck:是否自动确认
3. callback:回调对象
*/
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
/*
回调方法,当收到消息后,会自动执行该方法
1. consumerTag:标识
2. envelope:获取一些信息,交换机,路由key...
3. properties:配置信息
4. body:数据
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
/* System.out.println("consumerTag:"+consumerTag);
System.out.println("Exchange:"+envelope.getExchange());
System.out.println("RoutingKey:"+envelope.getRoutingKey());
System.out.println("properties:"+properties);*/
System.out.println("body:"+new String(body));
System.out.println("将日志信息打印到控制台.....");
}
};
channel.basicConsume(queue1Name,true,consumer);
//关闭资源?不要
}
}
接收端02:receive2.java
package test.rabbitmq.rabbitmq.mq5s.mq4;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class receive2 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//2. 设置参数
factory.setHost("127.0.0.1");//ip 默认值 localhost
factory.setPort(5672); //端口 默认值 5672
factory.setVirtualHost("/");//虚拟机 默认值/
factory.setUsername("guest");//用户名 默认 guest
factory.setPassword("guest");//密码 默认值 guest
//3. 创建连接 Connection
Connection connection = factory.newConnection();
//4. 创建Channel
Channel channel = connection.createChannel();
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
/*
basicConsume(String queue, boolean autoAck, Consumer callback)
参数:
1. queue:队列名称
2. autoAck:是否自动确认
3. callback:回调对象
*/
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
/*
回调方法,当收到消息后,会自动执行该方法
1. consumerTag:标识
2. envelope:获取一些信息,交换机,路由key...
3. properties:配置信息
4. body:数据
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
/* System.out.println("consumerTag:"+consumerTag);
System.out.println("Exchange:"+envelope.getExchange());
System.out.println("RoutingKey:"+envelope.getRoutingKey());
System.out.println("properties:"+properties);*/
System.out.println("body:"+new String(body));
System.out.println("将日志信息保存数据库.....");
}
};
channel.basicConsume(queue2Name,true,consumer);
//关闭资源?不要
}
}
模型图
发送端:Send.java
package test.rabbitmq.rabbitmq.mq5s.mq5;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Send {
private static final String EXCHANGE_NAME = "logs2";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
//debug日志
for (int i = 1; i <= 3; i++) {
String message = "debug_message" + i;
channel.basicPublish(EXCHANGE_NAME, "debug", null, message.getBytes());
}
//info日志
for (int i = 1; i <= 3; i++) {
String message = "info_message" + i;
channel.basicPublish(EXCHANGE_NAME, "info", null, message.getBytes());
}
//error日志
for (int i = 1; i <= 3; i++) {
String message = "error_message" + i;
channel.basicPublish(EXCHANGE_NAME, "error", null, message.getBytes());
}
channel.close();
connection.close();
}
}
接收端01:Receive1.java
package test.rabbitmq.rabbitmq.mq5s.mq5;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receive1 {
private static final String EXCHANGE_NAME = "logs2";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "debug");
Consumer consumer=new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println(message);
}
};
channel.basicConsume(queueName,true,consumer);
}
}
接收端02:Receive2.java
package test.rabbitmq.rabbitmq.mq5s.mq5;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receive2 {
private static final String EXCHANGE_NAME = "logs2";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "info");
DefaultConsumer consumer=new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println(message);
}
};
channel.basicConsume(queueName,true,consumer);
}
}
接收端03:Receive3.java
package test.rabbitmq.rabbitmq.mq5s.mq5;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Receive3 {
private static final String EXCHANGE_NAME = "logs2";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "error");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
FileUtil.save(message);
System.out.println("添加记录到文件!");
}
};
channel.basicConsume(queueName,true,consumer);
}
}
文章浏览阅读1.3k次。static: 是我们main函数的一种修饰符 静态修饰符:标识成员可以被类直接调用 static五种修饰1:静态成员变量2:静态方法变量3: 静态代码块4: 静态内部类5: 静态导包 package cn.biji; public class StatinTest { ..._java 中static中递归
文章浏览阅读729次。题目链接:http://abc048.contest.atcoder.jp/tasks/arc064_a?lang=enC - Boxes and CandiesTime limit : 2sec / Memory limit : 256MBScore : 300 pointsProblem StatementThere are N bo_atcoder arc064a
文章浏览阅读147次。[list][*][url=http://rikugun.iteye.com/blog/296384]半桶水开发 CMPP2.0 with Spring + JPA 1[/url][*][url=http://rikugun.iteye.com/blog/296439]半桶水开发 CMPP2.0 with Spring + JPA 2[/url][*][url=http://rik..._springboot cmpp2.0
文章浏览阅读6.9k次,点赞36次,收藏44次。使用Magisk Delta在各种模拟器安装Magisk到System分区的过程_magisk terminal emulator
文章浏览阅读1.7k次,点赞2次,收藏5次。目录1.date_format函数(根据格式整理日期)2.date_add函数(加减日期),date_sub,date_diff3.next_day函数4.last_day函数(求当月最后一天日期)1.date_format函数(根据格式整理日期)hive (gmall)> select date_format('2021-03-20','yyyy-MM');2020-03备注:与mysql中date_format函数的不同之处是:在hive中,可以指定为"yyyy-MM",在mysql中必_hive date format函数
文章浏览阅读1.6k次。最近sd卡读写出了问题,参考一篇网志解决。http://sns.linuxpk.com/space-1717-do-blog-id-15748.html 在embedded linux下插上一个U盘,在/dev/scsi/ 目录下,出现了4个part 。把该U盘插在pc机,在windows下_hc32f460 sd 卡 micro sd卡
文章浏览阅读1.2k次。当我们使用PC链接安卓手机的时候,如果手机没有开启usb调试模式,PC则不能够成功检测到我们的手机,有时候我们使用的一些功能强大的app好比以前我们使用的一个app引号精灵,老版本就需要打开usb调试模式下使用,现当新版本不需要了,因此我们需要找方法将手机的usb调试模式打开,以下内容我们记录一下华为畅玩7A AUM-AL00如何开启usb调试模式的步骤。接着,在华为畅玩7A AUM-AL00应用..._荣耀7a没有uab调试吗
文章浏览阅读1.1w次。(1)Advanced Science Letters(SCI)(2)Sensor Letters (SCI)(3)Computers & Electrical Engineering (SCI)(4)International Journal of Computational Intelligence Systems (SCI)(5)Future Generation Computer ..._cleaner environmental systems是sci吗
文章浏览阅读683次。json字符串中存在常规的用户输入的字符串,和很多的富文本样式标签(用户不能直接看到,点击富文本编辑器中的html源码按钮能看到),例如下面的:富文本<>sad<span>adzx我是用户输入富文本<>sad<span>adzx我是用户输入例如存在这种情况,需要把上面的这段字符串通过一个json格式,post到服务器端,假设这段字符串被赋给变量a :..._前端传递富文本数据需要注意什么
文章浏览阅读355次。时间限制:3000 ms | 内存限制:65535 KB难度:3 描述 现在,有一行括号序列,请你检查这行括号是否配对。 输入第一行输入一个数N(0<N<=100),表示有N组测试数据。后面的N行输入多组输入数据,每组输入数据都是一个字符串S(S的长度小于10000,且S不是空串),测试数据组数少于5组。数据保证S中只含有"[","]","(",")"四..._栈与括号配对
文章浏览阅读306次。EasyUI集成Kindeditor使用 在实际的项目中,我们需要在项目中集成富文本编辑器,而kindeditor作为一款优良的编辑器,在项目中或多或少都会用到!实际效果图 使用方法:1.首先下载Kindeditor编辑器,我这里使用的是4.1.10版本。下载地址:http://kind...
文章浏览阅读2.7k次。采取的方案node.js + Ffmpeg + jsmpeg工具node.js 下载路径https://pan.baidu.com/s/1DYnPW28hZz-I56jOopwxGQFfmpeg下载路径:https://pan.baidu.com/s/1KEGIYrRVLnLyDx1hwx4yBAjsmpeg下载路径:https://pan.baidu.com/s/1p5SnShAlTB..._html播放rtsp流