<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.20.0</version>
</dependency>
for (int i = 0; i < 10; i++) {
String mag = "hello work queue......" + i;
channel.basicPublish("", Constants.WORK_QUEUE, null, mag.getBytes());
}
Constant 包下 Constants 类
package rabbitmq.constant;
public class Constants {
public static final String HOST = "123.57.16.61";
public static final Integer PORT = 5672;
public static final String USERNAME = "study";
public static final String PASSWORD = "study";
public static final String VIRTUAL_HOST = "bite";
//声明一个工作队列
public static final String WORK_QUEUE = "work.queue";
}
完整代码:
package rabbitmq.work;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 工作队列模式
*/
public class Prooducer {
public static void main(String[] args) throws IOException, TimeoutException {
//创建一个连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(Constants.HOST);
connectionFactory.setPort(Constants.PORT);
connectionFactory.setUsername(Constants.USERNAME);
connectionFactory.setPassword(Constants.PASSWORD);
connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
//创建一个新的连接
Connection connection = connectionFactory.newConnection();
//开启一个通信
Channel channel = connection.createChannel();
//声明交换机,使用内置的交换机
//声明一个队列
channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);
//发送消息
for (int i = 0; i < 10; i++) {
String mag = "hello work queue......" + i;
channel.basicPublish("", Constants.WORK_QUEUE, null, mag.getBytes());
}
System.out.println("消息发送成功~");
//资源释放
channel.close();
connection.close();
}
}
package rabbitmq.work;
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer1 {
public static void main(String[] args) throws IOException, TimeoutException {
//创建一个连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(Constants.HOST);
connectionFactory.setPort(Constants.PORT);
connectionFactory.setUsername(Constants.USERNAME);
connectionFactory.setPassword(Constants.PASSWORD);
connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
//创建一个新的连接
Connection connection = connectionFactory.newConnection();
//开启一个通信
Channel channel = connection.createChannel();
//声明一个队列,如果队列不存在,则创建,如果队列存在,则不创建
channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);
//消费消息
//消费的逻辑
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("接受到消息:" + new String(body));
}
};
channel.basicConsume(Constants.WORK_QUEUE, true, consumer);
// channel.close();
// connection.close();
}
}
消费者2:
package rabbitmq.work;
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer2 {
public static void main(String[] args) throws IOException, TimeoutException {
//创建一个连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(Constants.HOST);
connectionFactory.setPort(Constants.PORT);
connectionFactory.setUsername(Constants.USERNAME);
connectionFactory.setPassword(Constants.PASSWORD);
connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
//创建一个新的连接
Connection connection = connectionFactory.newConnection();
//开启一个通信
Channel channel = connection.createChannel();
//声明一个队列,如果队列不存在,则创建,如果队列存在,则不创建
channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);
//消费消息
//消费的逻辑
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("接受到消息:" + new String(body));
}
};
channel.basicConsume(Constants.WORK_QUEUE, true, consumer);
// channel.close();
// connection.close();
}
}
本文系转载,版权归原作者所有,如若侵权请联系我们进行删除!
《数据资产管理白皮书》下载地址:https://www.dtstack.com/resources/1073/?src=bbs
《行业指标体系白皮书》下载地址:https://www.dtstack.com/resources/1057/?src=bbs
《数据治理行业实践白皮书》下载地址:https://www.dtstack.com/resources/1001/?src=bbs
《数栈V6.0产品白皮书》下载地址:https://www.dtstack.com/resources/1004/?src=bbs
想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友,浏览袋鼠云官网:https://www.dtstack.com/?src=bbs
同时,欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」,交流最新开源技术信息,群号码:30537511,项目地址:https://github.com/DTStack