package com.anqi.mq.nat;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class MyProducer {
  private static final String QUEUE_NAME = "ITEM_QUEUE";

  public static void main(String[] args) throws Exception {
    //1. 創(chuàng)建一個(gè) ConnectionFactory 并進(jìn)行設(shè)置
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    factory.setVirtualHost("/");
    factory.setUsername("guest");
    factory.setPassword("guest");

    //2. 通過連接工廠來創(chuàng)建連接
    Connection connection = factory.newConnection();

    //3. 通過 Connection 來創(chuàng)建 Channel
    Channel channel = connection.createChannel();

    //實(shí)際場(chǎng)景中，消息多為json格式的對(duì)象
    String msg = "hello";
    //4. 發(fā)送三條數(shù)據(jù)
    for (int i = 1; i <= 3 ; i++) {
      channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
      System.out.println("Send message" + i +" : " + msg);
    }

    //5. 關(guān)閉連接
    channel.close();
    connection.close();
  }
}

  /**
   * Declare a queue
   * @param queue the name of the queue
   * @param durable true if we are declaring a durable queue (the queue will survive a server restart)
   * @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
   * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
   * @param arguments other properties (construction arguments) for the queue
   * @return a declaration-confirm method to indicate the queue was successfully declared
   * @throws java.io.IOException if an error is encountered
   */
  Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments) throws IOException;

  /**
   * Publish a message
   * @see com.rabbitmq.client.AMQP.Basic.Publish
   * @param exchange the exchange to publish the message to
   * @param routingKey the routing key
   * @param props other properties for the message - routing headers etc
   * @param body the message body
   * @throws java.io.IOException if an error is encountered
   */
  void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;


  /**
   * Start a non-nolocal, non-exclusive consumer, with
   * a server-generated consumerTag.
   * @param queue the name of the queue
   * @param autoAck true if the server should consider messages
   * acknowledged once delivered; false if the server should expect
   * explicit acknowledgements
   * @param callback an interface to the consumer object
   * @return the consumerTag generated by the server
   * @throws java.io.IOException if an error is encountered
   * @see com.rabbitmq.client.AMQP.Basic.Consume
   * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
   * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
   */
  String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

package com.anqi.mq.nat;

import com.rabbitmq.client.*;
import java.io.IOException;

public class MyConsumer {

  private static final String QUEUE_NAME = "ITEM_QUEUE";

  public static void main(String[] args) throws Exception {
    //1. 創(chuàng)建一個(gè) ConnectionFactory 并進(jìn)行設(shè)置
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    factory.setVirtualHost("/");
    factory.setUsername("guest");
    factory.setPassword("guest");

    //2. 通過連接工廠來創(chuàng)建連接
    Connection connection = factory.newConnection();

    //3. 通過 Connection 來創(chuàng)建 Channel
    Channel channel = connection.createChannel();

    //4. 聲明一個(gè)隊(duì)列
    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

    /*
      true:表示自動(dòng)確認(rèn)，只要消息從隊(duì)列中獲取，無論消費(fèi)者獲取到消息后是否成功消費(fèi)，都會(huì)認(rèn)為消息已經(jīng)成功消費(fèi)
      false:表示手動(dòng)確認(rèn)，消費(fèi)者獲取消息后，服務(wù)器會(huì)將該消息標(biāo)記為不可用狀態(tài)，等待消費(fèi)者的反饋，如果消費(fèi)者一
      直沒有反饋，那么該消息將一直處于不可用狀態(tài)，并且服務(wù)器會(huì)認(rèn)為該消費(fèi)者已經(jīng)掛掉，不會(huì)再給其發(fā)送消息，
      直到該消費(fèi)者反饋。
    */

    //5. 創(chuàng)建消費(fèi)者并接收消息
    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(" [x] Received '" + message + "'");
      }
    };

    //6. 設(shè)置 Channel 消費(fèi)者綁定隊(duì)列
    channel.basicConsume(QUEUE_NAME, true, consumer);

  }
}

Send message1 : hello
Send message2 : hello
Send message3 : hello

 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'hello'
 [x] Received 'hello'
 [x] Received 'hello'

    <dependency>
      <groupId>com.rabbitmq</groupId>
      <artifactId>amqp-client</artifactId>
      <version>5.6.0</version>
    </dependency>    
        <dependency>
      <groupId>org.springframework.amqp</groupId>
      <artifactId>spring-rabbit</artifactId>
      <version>2.1.5.RELEASE</version>
    </dependency>

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xsi:schemaLocation="http://www.springframework.org/schema/rabbit
      https://www.springframework.org/schema/rabbit/spring-rabbit.xsd
      http://www.springframework.org/schema/beans
      https://www.springframework.org/schema/beans/spring-beans.xsd">

  <rabbit:connection-factory id="connectionFactory" host="localhost" virtual-host="/"
  username="guest" password="guest"/>
  <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"/>
  <rabbit:admin connection-factory="connectionFactory"/>
  <rabbit:queue name="MY-QUEUE"/>
</beans>

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class Main {
  public static void main(String[] args) {
    ApplicationContext app = new ClassPathXmlApplicationContext("spring/rabbit-context.xml");
    AmqpTemplate amqpTemplate = app.getBean(AmqpTemplate.class);
    amqpTemplate.convertAndSend("MY-QUEUE", "Item");
    String msg = (String) amqpTemplate.receiveAndConvert("MY-QUEUE");
    System.out.println(msg);
  }
}

/**
 * Convert a Java object to an Amqp {@link Message} and send it to a specific exchange
 * with a specific routing key.
 *
 * @param exchange the name of the exchange
 * @param routingKey the routing key
 * @param message a message to send
 * @throws AmqpException if there is a problem
 */
void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException;
/**
   * Receive a message if there is one from a specific queue and convert it to a Java
   * object. Returns immediately, possibly with a null value.
   *
   * @param queueName the name of the queue to poll
   * @return a message or null if there is none waiting
   * @throws AmqpException if there is a problem
   */
@Nullable
Object receiveAndConvert(String queueName) throws AmqpException;