Rabbit简单队列模式

 1 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.kf</groupId>
<artifactId>rabbitMQ.demo</artifactId>
<version>0.0.1-SNAPSHOT</version> <dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.6.5</version>
</dependency>
</dependencies> </project>
 package com.kf.utils;

 import java.io.IOException;
import java.util.concurrent.TimeoutException; import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory; /**
* 获取rabbit
* @author kf
*
*/
public class RabbitConnectionUtils { public static Connection getConnection() throws IOException, TimeoutException{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
factory.setUsername("admin");
factory.setPassword("admin");
//AMQP协议端口号
factory.setPort(5672);
factory.setVirtualHost("/kf");
Connection newConnection = factory.newConnection();
return newConnection;
} }
 package com.kf.queueDemo.simpleQueue;

 import java.io.IOException;
import java.util.concurrent.TimeoutException; import com.kf.utils.RabbitConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection; /**
* 简单队列
* @author kf
*
*/
public class SimpleQueueProducer { //队列名称
private static String QUEUENAME = "SIMPLEQUEUE"; public static void main(String[] args) throws IOException, TimeoutException{
Connection connection = RabbitConnectionUtils.getConnection(); //创建通道
Channel channel = connection.createChannel(); //通道里放入队列
/**
* 第一个参数是 队列名称
* 第二个参数指 要不要持久化
*/
channel.queueDeclare(QUEUENAME, false, false, false, null); /* //消息体
String mes = "demo_message汉字"; //发送消息
*//**
* 参数为 exchange, routingKey, props, body
* exchange 交换机
* routingKey 路由键
*
* body 消息体
*//*
channel.basicPublish("", QUEUENAME, null, mes.getBytes());*/ /**
* 集群环境下,多个消费者情况下。消费者默认采用均摊
*/
for(int i=1; i<11; i++){
String mes = "demo_message汉字"+i;
System.out.println("发送消息"+mes);
channel.basicPublish("", QUEUENAME, null, mes.getBytes());
} // System.out.println("发送消息"+mes); channel.close();
connection.close();
} }
 package com.kf.queueDemo.simpleQueue;

 import java.io.IOException;
import java.util.concurrent.TimeoutException; import com.kf.utils.RabbitConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties; /**
* 简单队列消费者
* @author kf
*
*/
public class SimpleConsumer {
//队列名称
private static String QUEUENAME = "SIMPLEQUEUE"; public static void main(String[] args) throws IOException, TimeoutException{
System.out.println("01开始接收消息");
Connection connection = RabbitConnectionUtils.getConnection(); //创建通道
final Channel channel = connection.createChannel(); //通道里放入队列
/**
* 第一个参数是 队列名称
* 第二个参数指 要不要持久化
*/
channel.queueDeclare(QUEUENAME, false, false, false, null); DefaultConsumer consumer = new DefaultConsumer(channel){
//监听队列
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
System.out.println("------------进入监听---------");
String s = new String(body, "utf-8");
System.out.println("获取到的消息是:"+s);
//手动应答。
/**
* 当 channel.basicConsume(QUEUENAME, true, consumer);第二个参数为false时 是手动应答模式
*/
// channel.basicAck(envelope.getDeliveryTag(), false);
}
}; //设置应答模式
/**
* 参数: 对列名,是否自动签收,监听的类
*/
System.out.println("获取消息的方法之前");
channel.basicConsume(QUEUENAME, true, consumer);
System.out.println("获取消息的方法之后"); } }
上一篇:C# StreamReader对象


下一篇:C#读写Shapefile