package com.rabbitmq.demo;
import com.rabbitmq.client.*;
public class Produce {
public static void main(String args[]) throws Exception{
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setPassword("guest");
connectionFactory.setPassword("guest");
connectionFactory.setHost("192.168.1.141");
connectionFactory.setPort(5672);
Connection connection=connectionFactory.newConnection();
Channel channel= connection.createChannel();
String exchangeName="test";
String queueName="test";
String routingkey="routingkey";
String bindingkey=routingkey;
//设置交换机,直连，持久化，非自动删除，参数
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,null);
//channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,null);
/**
* BuiltinExchangeType.FANOUT 不需要绑定 bindingkey与routingkey exchange直接将消息投递到queue
* BuiltinExchangeType.DIRECT 需要绑定 且bindingkey与routingkey一致时 exchange投递消息到queue
* BuiltinExchangeType.TOPIC 需要绑定 模糊匹配,规则如下
* RoutingKey 为一个点号"."分隔的字符串(被点号"。"分隔开的每段独立的字符,如com.rabbit.demo,com.rabbit.test）
* BindingKey RoutingKey 样也是点号"."分隔的字符串;
* BindingKey 中可以存在两种特殊字符串"*"和"#"，用于做模糊匹配，其中"#"用于匹配多个（零个到多个），* 匹配任意一个单词。
* 如果binding_key 是 “#” - 它会接收所有的Message，不管routing_key是什么，就像是fanout exchange。
* 如果 “*” and “#” 没有被使用，那么topic exchange就变成了direct exchange。
* BuiltinExchangeType.HEADERS 不需要绑定
* 消费者arguments指定“x-match”，这个键的Value可以是any或者all，这代表消息携带的Hash是需要全部匹配(all)，还是仅匹配一个键(any)就可以了
*/
/**
* 持久化:数据会保存到磁盘，重启rabbitmq数据依旧存在
*
* 排他:该队列仅对首次声明它的连接可见，并在连接断开时自动删除。
* 需要注意三点:排他队列是基于连接( Connection) 可见的，同一个连接的不同信道 (Channel)
* 是可以同时访问同一连接创建的排他队列; "首次"是指如果一个连接己经声明了
* 排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同:即使该队
* 列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除，这种队列
* 适用于一个客户端同时发送和读取消息的应用场景。
*
*
* 自动删除:设置是否自动删除。为 true 则设置队列为自动删除。自动删除的前提是:
* 至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会
* 自动删除。不能把这个参数错误地理解为: "当连接到此队列的所有客户端断开时，这
* 个队列自动删除"，因为生产者客户端创建这个队列，或者没有消费者客户端与这个队
* 列连接时，都不会自动删除这个队列。
*
* 根据业务数据，最好提前建好exchange，queue,及绑定关系，生产端，消费端可以避免很多错误,如exchang创建失败，
* 绑定关系不确定导致消息投递失败
*
* rabbitmq clinet 属性集
* props 消息的基本属性集，其包含 14 个属性成员，分别有 contentType
* content ncoding headers Map<String Object>) deliveryMode priority
* correlationld replyTo expiration messageld timestamp type userld
* appld cluster 。
*
* mandatory 参数设为 true 时，交换器无法根据自身的类型和路由键的队列，
* 那么 RabbitM 会调用 Basic.Return 命令将消息返回给生产者 数设置为 false 时，出现上述情形，
* 则消息直接被丢弃 那么生产者如何获取到没有被正确路由到合适队列的消息呢?这时channel addReturnListener 来添加 ReturnListener 监昕器实现。
* 通过rabbitmq备份队列，当消息发送不到主队列，将自动发送到备份队列，设置了备份队列，mandatory参数将无效
*/
//设置队列，持久化，非排他，非自动删除
channel.queueDeclare(queueName,true,false,false,null);
//channel.queueBind(queueName,exchangeName, bindingkey);
for (int i = 0; i <10 ; i++) {
channel.basicPublish(exchangeName,routingkey,
MessageProperties.PERSISTENT_TEXT_PLAIN,"hi,Rabbitmq".getBytes() );
}
channel.close();
connection.close();
}
}
package com.rabbitmq.demo;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer {
public static void main(java.lang.String args[]) throws Exception{
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setPassword("guest");
connectionFactory.setPassword("guest");
connectionFactory.setHost("192.168.1.141");
connectionFactory.setPort(5672);
Connection connection=connectionFactory.newConnection();
Channel channel= connection.createChannel();
/*while(true){
//单个获取消息，当没有消息返回null，pull模式
GetResponse response = channel.basicGet("test001", true);
if(response!=null){
System.out.println(new String(response.getBody()));
}
}*/
//从queue中手动确认获取消息，push模式
/**
* 队列名，自动签收，默认消费者
* queue 队列的名称:
* autoAck 设置是否自动确认。建议设成 fa se ，即不自动确认:
* utoAck 参数置为 false ，对于 RabbitMQ 服务端而 ，队列中的消息分成了两个部分
* 部分是等待投递给消费者的消息:一部分是己经投递给消费者，但是还没有收到消费者确认
* 信号的消息。 如果 RabbitMQ 直没有收到消费者的确认信号，并且消费此消息的消费者己经
* 断开连接，则 RabbitMQ 会安排该消息重新进入队列，等待投递给下 个消费者，当然也有可
* 能还是原来的那个消费者。
* RabbitMQ 会为未确认的消息设置过期时间，它判断此消息是否需要重新投递给消费者的
* 依据是消费该消息的消费者连接是否己经断开，这么设计的原因是 RabbitMQ 允许消费者
* 消费 条消息的时间可以很久很久。
* consumerTag: 消费者标签，用来区分多个消费者:
* noLocal 设置为 true 则表示不能将同一个 Connectio口中生产者发送的消息传送给
* 这个 Connection 中的消费者:
* exclusive 设置是否排他
* arguments 设置消费者的其他参数:
* callback 设置消费者的回调函数。用来处理 Rabb itM 推送过来的消息，比如
* DefaultConsumer 使用时需要客户端重写 (overr e) 其中的方法。
*/
channel.basicConsume("test",false,new DefaultConsumer(channel){
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1:"+new java.lang.String(body));
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
//创建多个消费者，默认为平均分摊round-robin
//不支持队列层面的广播消费
/**
* 上面代码中显式地设置 autoAck false 然后在接收到消息之后 行显式 ack
* (channel basicAck )， 对于消费者来说这 设置是非常 可以防止
* 丢失。
*
*
*/
channel.basicConsume("queue",false,new DefaultConsumer(channel){
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者2："+new java.lang.String(body));
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
//单个拒绝接受消息
//channel.basicReject(111, false);
/**
* 多个拒绝消息
*/
// channel.basicNack();
/**
* channel.basicReject 或者 channel.basicNack 中的 requeue 设直为 false ，可
* 以启用"死信队列"的功能。死信队列可以通过检测被拒绝或者未送达的消息来追踪问题
*/
/***
* Basic Consume 将信道 (Channel) 直为接收模式，直到取消队列的订阅为止。在接收
* 模式期间， RabbitMQ 会不断地推送消息给消费者，当然推送消息的个数还是会受到 Basic.Qos
* 的限制.如果只想从队列获得单条消息而不是持续订阅，建议还是使用 Basic.Get 进行消费.但
* 是不能将 Basic.Get 放在一个循环里来代替 Basic.Consume ，这样做会严重影响 RabbitMQ
* 的性能.如果要实现高吞吐量，消费者理应使用 Basic.Consume 方法。
*
*
*/
}
}