Rabbitmq基本API使⽤⼀、⽣产者1. 创建ConnectionFactory⼯⼚（地址、⽤户名、密码、vhost）2. 创建Connection3. 创建信道（Channel）4. 创建 exchange（指定名称、类型-DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");、是否持久化）5. 发送消息（指定：exchange、发送的routingKey ，发送到的消息）基础的⽣产者：public class TestProducer {public final static String EXCHANGE_NAME = "direct_logs";public static void main(String[] args)throws IOException, TimeoutException {/* 创建连接,连接到RabbitMQ*/ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.112.131");connectionFactory.setVirtualHost("my_vhost");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = connectionFactory.newConnection();/*创建信道*/Channel channel = connection.createChannel();/*创建交换器*/channel.exchangeDeclare(EXCHANGE_NAME,"direct");//channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT);/*⽇志消息级别，作为路由键使⽤*/String[] routekeys = {"king","queue","prince"};for(int i=0;i<3;i++){String routekey = routekeys[i%3];String msg = "Hellol,RabbitMq"+(i+1);/*发布消息，需要参数：交换器，路由键，其中以⽇志消息级别为路由键*/channel.basicPublish(EXCHANGE_NAME,routekey,null,msg.getBytes());System.out.println("Sent "+routekey+":"+msg);}channel.close();connection.close();}}View Code⼆、消费者1. 创建ConnectionFactory⼯⼚（地址、⽤户名、密码、vhost）2. 创建Connection3. 创建信道（Channel）4. 声明⼀个 exchange（指定名称、类型、是否持久化）5. 创建⼀个队列（指定：名称，是否持久化，是否独占，是否⾃动删除，其他参数）6. 队列、exchange通过routeKey进⾏绑定7. 消费者接收消息（队列名称，是否⾃动ACK）基本的消费者：public class TestConsumer {public static void main(String[] argv)throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.112.131");factory.setVirtualHost("my_vhost");factory.setUsername("admin");factory.setPassword("admin");// 打开连接和创建频道，与发送端⼀样Connection connection = factory.newConnection();final Channel channel = connection.createChannel();channel.exchangeDeclare(TestProducer.EXCHANGE_NAME,"direct");/*声明⼀个队列*/String queueName = "focuserror";channel.queueDeclare(queueName,false,false,false,null);/*绑定，将队列和交换器通过路由键进⾏绑定*/String routekey = "king";/*表⽰只关注error级别的⽇志消息*/channel.queueBind(queueName,TestProducer.EXCHANGE_NAME,routekey);System.out.println("waiting for message........");/*声明了⼀个消费者*/final Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("Received["+envelope.getRoutingKey()+"]"+message);}};/*消费者正式开始在指定队列上消费消息*/channel.basicConsume(queueName,true,consumer);}}View Code三、消息持久化1. exchange 需要持久化2. 发送消息设置参数为 MessageProperties.PERSISTENT_TEXT_PLAIN3. 队列需要设置参数为持久化1、//TODO 创建持久化交换器 durable=truechannel.exchangeDeclare(EXCHANGE_NAME,"direct",true);2、//TODO 发布持久化的消息(delivery-mode=2)channel.basicPublish(EXCHANGE_NAME,routekey,MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());3、//TODO 声明⼀个持久化队列(durable=true)// autoDelete=true 消费者停⽌了，则队列会⾃动删除//exclusive=true独占队列，只能有⼀个消费者消费String queueName = "msgdurable";channel.queueDeclare(queueName,true,false,false,null);四、如何⽀持事务（防⽌投递消息的时候消息丢失-效率特别低，不建议使⽤，可以使⽤⽣产者ACK机制）1. 启动事务2. 成功提交3. 失败则回滚//TODO//加⼊事务channel.txSelect();try {for(int i=0;i<3;i++){String routekey = routekeys[i%3];// 发送的消息String message = "Hello World_"+(i+1)+("_"+System.currentTimeMillis());channel.basicPublish(EXCHANGE_NAME, routekey, true,null, message.getBytes());System.out.println("----------------------------------");System.out.println(" Sent Message: [" + routekey +"]:'"+ message + "'");Thread.sleep(200);}//TODO//事务提交channel.txCommit();} catch (IOException e) {e.printStackTrace();//TODO//事务回滚channel.txRollback();} catch (InterruptedException e) {e.printStackTrace();}View Code五、消费消息⼿动ACK，如果异常则使⽤拒绝的⽅式，然后异常消息推送到-死信队列批量ack的时候如果其中有⼀个消息出现异常，则会导致消息丢失（⽇志处理的时候可以使⽤批量）1 /*消费者正式开始在指定队列上消费消息，第⼆个参数false为⼿动应答*/channel.basicConsume(queueName,false,consumer);2 收到消息以后，⼿动应答数据接收成功channel.basicAck(envelope.getDeliveryTag(),false);3 收到消息，如果处理失败则拒绝消息：DeliveryTag是消息在队列中的标识channel.basicReject(envelope.getDeliveryTag(),false);4 决绝的参数说明//TODO Reject⽅式拒绝(这⾥第2个参数决定是否重新投递),不要重复投递，因为消息重复投递后处理可能依然异常//channel.basicReject(envelope.getDeliveryTag(),false);//TODO Nack⽅式的拒绝（第2个参数决定是否批量,第3个参数是否重新投递）channel.basicNack(envelope.getDeliveryTag(), false, true);View Code六、创建队列的参数解析：场景，延迟队列，保存带有时效性的订单，⼀旦订单过期，则信息会转移到死信队列//TODO /*⾃动过期队列--参数需要Map传递*/String queueName = "setQueue";Map<String, Object> arguments = new HashMap<String, Object>();arguments.put("x-expires",10*1000);//消息在队列中保存10秒后被删除//TODO 队列的各种参数/*加⼊队列的各种参数*/// autoDelete=true 消费者停⽌了，则队列会⾃动删除//exclusive=true独占队列，只能有⼀个消费者消费channel.queueDeclare(queueName,true,true, false,arguments);七、发送消息以后带有应答的队列1. 声明⼀个回应队列2. 声明⼀个回应消息的消费者3. 声明⼀个属性对象（指定队列，会唯⼀的id）4. ⽣产者发送消息给消费者（带着回应队列）5. 消费者接收到消息以后根据对应的信息，给予回应⽣产者端：1、//TODO 响应QueueName ，消费者将会把要返回的信息发送到该QueueString responseQueue = channel.queueDeclare().getQueue();//TODO 消息的唯⼀idString msgId = UUID.randomUUID().toString();2、/*声明了⼀个消费者*/final Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("Received["+envelope.getRoutingKey()+"]"+message);}};//TODO 消费者应答队列上的消息channel.basicConsume(responseQueue,true,consumer);3、//TODO 设置消息中的应答属性AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().replyTo(responseQueue).messageId(msgId).build();4、String msg = "Hello,RabbitMq";//TODO 发送消息时，把响应相关属性设置进去channel.basicPublish(EXCHANGE_NAME,"error",properties,msg.getBytes());View Code消费者端：String message = new String(body, "UTF-8");System.out.println("Received["+envelope.getRoutingKey()+"]"+message);//TODO 从消息中拿到相关属性（确定要应答的消息ID,）AMQP.BasicProperties respProp= new AMQP.BasicProperties.Builder().replyTo(properties.getReplyTo()).correlationId(properties.getMessageId()).build();//TODO 消息消费时，同时需要⽣作为⽣产者⽣产消息（以OK为标识）channel.basicPublish("", respProp.getReplyTo() ,respProp ,("OK,"+message).getBytes("UTF-8"));⼋、死信队列 - 下列消息会放到死信队列1. 消息被否定确认，使⽤ channel.basicNack 或 channel.basicReject ，并且此时requeue 属性被设置为false。
