Spring AMQP简介和使用

public class Message {
    private final MessageProperties messageProperties;
    private final byte[] body;

    public Message(byte[] body, MessageProperties messageProperties) {
        this.body = body;
        this.messageProperties = messageProperties;
    }

    public byte[] getBody() {
        return this.body;
    }

    public MessageProperties getMessageProperties() {
        return this.messageProperties;
    }
}

public interface Exchange {
    String getName();
    String getExchangeType();
    boolean isDurable();
    boolean isAutoDelete();
    Map<String, Object> getArguments();
}

public class Queue  {
    private final String name;
    private volatile boolean durable;
    private volatile boolean exclusive;
    private volatile boolean autoDelete;
    private volatile Map<String, Object> arguments;

    /**
     * The queue is durable, non-exclusive and non auto-delete.
     *
     * @param name the name of the queue.
     */
    public Queue(String name) {
        this(name, true, false, false);
    }

    public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete) {
    this(name, durable, exclusive, autoDelete, null);
    }

    // Getters and Setters omitted for brevity
}

public class Binding extends AbstractDeclarable {
    public enum DestinationType {
        QUEUE, EXCHANGE;
    }

    private final String destination;
    private final String exchange;
    private final String routingKey;
    private final Map<String, Object> arguments;
    private final DestinationType destinationType;

    public Binding(String destination, DestinationType destinationType, String exchange, String routingKey,
            Map<String, Object> arguments) {
        this.destination = destination;
        this.destinationType = destinationType;
        this.exchange = exchange;
        this.routingKey = routingKey;
        this.arguments = arguments;
    }
}

``new Binding(someQueue, someDirectExchange, "foo.bar")``

``new Binding(someQueue, someTopicExchange, "foo.*")``

``new Binding（someQueue，someFanoutExchange）``

``Binding b = BindingBuilder.bind(someQueue).to(someTopicExchange).with("foo.*");``

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-amqp</artifactId>
<version>1.7.6.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.7.6.RELEASE</version>
</dependency>

public static void main(String[] args) {
    CachingConnectionFactory connectionFactory = new CachingConnectionFactory("192.168.0.27", 5672);
    connectionFactory.setUsername("admin");
    connectionFactory.setPassword("123456");
    // 通过AmqpAdmin来申明队列
    AmqpAdmin admin = new RabbitAdmin(connectionFactory);
    // 匿名队列，默认的命名为UUID
    Queue queue = new AnonymousQueue();
    // 使用Base64命名策略，添加自定义前缀
    // Queue queue = new AnonymousQueue(new AnonymousQueue.Base64UrlNamingStrategy("demo-"));
    admin.declareQueue(queue);
    System.err.println("Queue name is : " + queue.getName());
    AmqpTemplate template = new RabbitTemplate(connectionFactory);
    String str = "this is foo.";
    template.convertAndSend(queue.getName(), str);
    System.err.println("Send : " + str);
    String foo = (String) template.receiveAndConvert(queue.getName());
    System.err.println("Received : " + foo);
    User user = new User("李四");
    System.err.println("Send : " + user);
    template.convertAndSend(queue.getName(), user);
    // 接收消息并直接转换为对象，最大等待2秒
    user = (User) template.receiveAndConvert(queue.getName(), 1000 * 2);
    System.err.println("Received : " + user);
}

public class User implements Serializable {
    private String name;

    public User() {
    }

    public User(String name) {
        this.name = name;
    }

    // 省略getter、setter

    @Override
    public String toString() {
        return "User{" +
                "name='" + name + '\'' +
                '}';
    }
]

public class ReceiveAndReplyDemo {
    public static void main(String[] args) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory("192.168.0.27", 5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("123456");

        // 通过AmqpAdmin来申明队列
        AmqpAdmin admin = new RabbitAdmin(connectionFactory);
        Queue queue = new AnonymousQueue();
        admin.declareQueue(queue);

        RabbitTemplate template = new RabbitTemplate(connectionFactory);

        // 发送消息，生产者
        template.convertAndSend(queue.getName(), new User("王五"));
        // 接收消息并直接转换为对象，并进行回复，消费者
        boolean received = template.receiveAndReply(queue.getName(),
                // 回复回调
                new ReceiveAndReplyCallback<User, User>() {
                    @Override
                    public User handle(User payload) {
                        System.err.println("Handle : " + payload);
                        // 将消息payload修改改一下
                        payload.setName("王五改名了");
                        return payload;
                    }
                },
                // 设置回复地址
                new ReplyToAddressCallback<User>() {
                    @Override
                    public Address getReplyToAddress(Message request, User reply) {
                        // 改完后的消息回复给发送方
                        System.err.println("Replying : " + reply);
                        return new Address("", queue.getName());
                    }
                }
        );

        // 收到消费者的回复，生产者
        User replyUser = (User) template.receiveAndConvert(queue.getName(), 1000 * 3);
        System.err.println("Replied user : " + replyUser);

        System.err.println("Received : " + received);
    }
}

public class ConfirmAndReturnDemo {
    public static void main(String[] args) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory("192.168.0.27", 5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("123456");

        // 开启消息确认和回调
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);

        Queue queue = new AnonymousQueue();
        AmqpAdmin admin = new RabbitAdmin(connectionFactory);
        admin.declareQueue(queue);

        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        // 设置为true，则消息改为手动返回
        template.setMandatory(true);
        // 设置消息返回回调，一个RabbitTemplate只能设置一次返回回调
        // 当消息不能成功投递，会抛出AmqpMessageReturnedException，该异常包含ReturnCallback所需的参数信息，此时会执行回调
        template.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            System.err.println("Return callback : ");
            System.err.println("    message : " + message);
            System.err.println("    reply   : " + replyCode + ", " + replyText);
        });
        // 消息确认回调，一个RabbitTemplate只能设置一次确认回调
        template.setConfirmCallback((correlationData, ack, cause) -> {
            System.err.println("Confirm callback : ");
            System.err.println("    correlationData : " + correlationData);
            System.err.println("    ack   : " + ack);
            System.err.println("    cause : " + cause);
        });

        String str = "this is foo.";
        template.convertAndSend(queue.getName(), (Object) str, new CorrelationData(UUID.randomUUID().toString()));
        System.err.println("Send : " + str);
        String foo = (String) template.receiveAndConvert(queue.getName());
        System.err.println("Received : " + foo);

        User user = new User("zhangsan");
        template.convertAndSend(queue.getName(), user, new CorrelationData(UUID.randomUUID().toString()));
        System.err.println("Send : " + user);
        User receivedUser = (User) template.receiveAndConvert(queue.getName());
        System.err.println("Received : " + receivedUser);

        // 不能成功投递消息
        System.err.println("Mock message body can't be sent.");
        str = "can not be sent.";
        // routing key无法匹配，则confirmCallback和returnCallback都会被触发，ack仍然为true
        template.convertAndSend("dontExist.exchange", (Object) str, new CorrelationData(UUID.randomUUID().toString()));
        // exchange找不到，则会触发confirmCallback，但是returnCallback不会被触发，ack为false，cause包含异常信息，此时channel会关闭
        template.convertAndSend("dontExist.exchange", "dontExist.routingKey", str, new CorrelationData(UUID.randomUUID().toString()));
    }
}

Send : this is foo.
Confirm callback :
    correlationData : CorrelationData [id=71ec4137-8fe6-43dd-9552-147a7807b686]
    ack   : true
    cause : null
Received : this is foo.

Confirm callback :
    correlationData : CorrelationData [id=662ef2e6-41f9-4461-8c7f-d654ab463ded]
    ack   : true
Send : User{name='zhangsan'}
    cause : null
Received : User{name='zhangsan'}

Mock message body can't be sent.
Return callback :
    message : (Body:'can not be sent.' MessageProperties [headers={}, timestamp=null, ……
    reply   : 312, NO_ROUTE
Confirm callback :
    correlationData : CorrelationData [id=23386712-9922-4682-b92f-70e42bb28182]
    ack   : true
    cause : null

Confirm callback :
    correlationData : CorrelationData [id=bbdb39ed-5ad8-4818-bd96-eab58c8d5dad]
    ack   : false
    cause : channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'cant_sent_exchange' in vhost '/', class-id=60, method-id=40)