RabbitMQ

安装

docker

docker pull rabbitmq
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 -v /data/rabbimq:/var/lib/rabbitmq --hostname myRabbit -e RABBITMQ_DEFAULT_VHOST=my_vhost  -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin rabbitmq
docker exec -it rabbit rabbitmq-plugins enable rabbitmq_management

-d 后台运行容器；
—name 指定容器名；
-p 指定服务运行的端口（5672：应用访问端口；15672：控制台Web端口号）；
-v 映射目录或文件；
—hostname 主机名（RabbitMQ的一个重要注意事项是它根据所谓的 “节点名称” 存储数据，默认为主机名）；
-e 指定环境变量；（RABBITMQ_DEFAULT_VHOST：默认虚拟机名；RABBITMQ_DEFAULT_USER：默认的用户名；RABBITMQ_DEFAULT_PASS：默认用户名的密码）

延迟队列

package com.huiyou.washing;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
/**
 * Created by hidden on 2017/2/7.
 */
public class DelayTest {
    public static final String ip = "192.168.31.232";
    public static final int port = 5672;
    public static final String username = "admin";
    public static final String password = "rabbit_0921";
    public static final String queueName = "queue.ttl.test";
    public static final String exchangeName = "exchange.ttl.test";
    public static final String routingKey = "ttl";
    public static final Boolean durable = true;
    public static final Boolean exclusive = false;
    public static final Boolean autoDelete = false;
    public static void main(String[] args) {
        try {
            Connection connection = getConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare("dlx.exchange", "direct");
            channel.queueDeclare("dlx.queue", true, false, false, null);
            channel.queueBind("dlx.queue", "dlx.exchange", "dlx.routingKey");
            //创建测试超时的Exchange及Queue
            channel.exchangeDeclare("delay.exchange", "direct");
            Map<String, Object> arguments = new HashMap<>();
            //过期时间10s
            arguments.put("x-message-ttl", 10000);
            //绑定DLX
            arguments.put("x-dead-letter-exchange", "dlx.exchange");
            //绑定发送到DLX的RoutingKey
            arguments.put("x-dead-letter-routing-key", "dlx.routingKey");
            channel.queueDeclare("delay.queue", true, false, false, arguments);
            channel.queueBind("delay.queue", "delay.exchange", "delay.routingKey");
            //发布一条消息
            channel.basicPublish("delay.exchange", "delay.routingKey", null, "该消息将在10s后发送到延迟队列".getBytes());
            channel.close();
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
    private static Connection getConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(ip);
        factory.setPort(port);
        factory.setUsername(username);
        factory.setPassword(password);
        return factory.newConnection();
    }
}