rabbitMQ

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

public class Producer {
    //队列
    private static final String QUEUE = "helloworld";
    public static void main(String[] args) throws Exception {
        //通过连接工厂创建新的连接和mq建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.33.10");
        connectionFactory.setPort(5672);//端口
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        //设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
        connectionFactory.setVirtualHost("/");

        Connection connection = null;
        Channel channel = null;
        //建立新连接
        connection = connectionFactory.newConnection();
        //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
        channel = connection.createChannel();
        //声明队列，如果队列在mq 中没有则要创建
        //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
        /**
         * 参数明细
         * 1、queue 队列名称
         * 2、durable 是否持久化，如果持久化，mq重启后队列还在
         * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
         * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
         * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
         */
        channel.queueDeclare(QUEUE, true, false, false, null);
        //发送消息
        //参数：String exchange, String routingKey, BasicProperties props, byte[] body
        /**
         * 参数明细：
         * 1、exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）
         * 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称
         * 3、props，消息的属性
         * 4、body，消息内容
         */
        //消息内容
        String message = "hello world";
        channel.basicPublish("", QUEUE, null, message.getBytes());
        System.out.println("send to mq " + message);
        //关闭连接，//先关闭通道
        channel.close();
        connection.close();

    }
}

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {
    //队列
    private static final String QUEUE = "helloworld";

    public static void main(String[] args) throws IOException, TimeoutException {
        //通过连接工厂创建新的连接和mq建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.33.10");
        connectionFactory.setPort(5672);//端口
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
        connectionFactory.setVirtualHost("/");

        //建立新连接
        Connection connection = connectionFactory.newConnection();
        //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
        Channel channel = connection.createChannel();

        //监听队列
        //声明队列，如果队列在mq 中没有则要创建
        //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
        /**
         * 参数明细
         * 1、queue 队列名称
         * 2、durable 是否持久化，如果持久化，mq重启后队列还在
         * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
         * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
         * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
         */
        channel.queueDeclare(QUEUE,true,false,false,null);

        //实现消费方法
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){

            /**
             * 当接收到消息后此方法将被调用
             * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
             * @param envelope 信封，通过envelope
             * @param properties 消息属性
             * @param body 消息内容
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //交换机
                String exchange = envelope.getExchange();
                //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
                long deliveryTag = envelope.getDeliveryTag();
                //消息内容
                String message= new String(body,"utf-8");
                System.out.println("receive message:"+message);
            }
        };

        //监听队列
        //参数：String queue, boolean autoAck, Consumer callback
        /**
         * 参数明细：
         * 1、queue 队列名称
         * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
         * 3、callback，消费方法，当消费者接收到消息要执行的方法
         */
        channel.basicConsume(QUEUE,true,defaultConsumer);

    }
}

@RestController
public class Producer {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @RequestMapping("/simpleQueue")
    public String testSimpleQueue(){
        String queueName ="helloworld";
        String message = "hello,simple";
        rabbitTemplate.convertAndSend(queueName,message);
        return "success";
    }
}

@Component
public class Consumer {
    @RabbitListener(queues = "helloworld")
    public  void  listenSimpleQueue(String msg) throws InterruptedException {
        System.out.println("接收的消息：" + msg);
    }
}

@Configuration
public class FanoutConfig {
    //声明交换机
    @Bean(name="fanoutExchange")
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanout.exchange");
    }

    //声明队列
    @Bean(name="fanoutQueue1")
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    @Bean
    public Binding bindingQueue1(@Qualifier("fanoutQueue1") Queue queue
            ,@Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }

    //声明队列
    @Bean(name="fanoutQueue2")
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
    @Bean
    public Binding bindingQueue2(@Qualifier("fanoutQueue2") Queue queue
            ,@Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }
}

@RabbitListener(queues = "fanout.queue1")
public  void  listenFanoutQueue1(String msg) throws InterruptedException {
	System.out.println("消费者1接收消息：" + msg);
	Thread.sleep(20);
}

@RabbitListener(queues = "fanout.queue2")
public  void  listenFanoutQueue2(String msg) throws InterruptedException {
	System.out.println("消费者2接收消息：" + msg);
	Thread.sleep(50);
}

@RabbitListener(bindings = @QueueBinding(
            value=@Queue(name = "direct.queue1"),
            exchange = @Exchange(name="direct.exchange",type = ExchangeTypes.DIRECT),
            key={"eat","jump"}
    ))
    public  void  listenRouteQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1--Direct接收消息--eat：" + msg);
        Thread.sleep(20);
    }

    @RabbitListener(bindings = @QueueBinding(
            value=@Queue(name = "direct.queue2"),
            exchange = @Exchange(name="direct.exchange",type = ExchangeTypes.DIRECT),
            key={"jump"}
    ))
    public  void  listenRouteQueue2(String msg) throws InterruptedException {
        System.out.println("消费者1--Direct接收消息--jump：" + msg);
        Thread.sleep(20);
    }

@RequestMapping("/routeQueue")
public String testRouteQueue() throws InterruptedException {
    String exchange = "direct.exchange";
    //交换机，路由名，信息
    rabbitTemplate.convertAndSend(exchange,"eat","...eat.....");
    rabbitTemplate.convertAndSend(exchange,"jump","...jump.....");

    return "routeQueue success";
}

@RabbitListener(bindings = @QueueBinding(
            value=@Queue(name = "topic.queue1"),
            exchange = @Exchange(name="topic.exchange",type = ExchangeTypes.TOPIC),
            key={"wanho.*"}
    ))
    public  void  listenTopicQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1--topic接收消息--wanho：" + msg);
        Thread.sleep(20);
    }
    @RabbitListener(bindings = @QueueBinding(
            value=@Queue(name = "topic.queue2"),
            exchange = @Exchange(name="topic.exchange",type = ExchangeTypes.TOPIC),
            key={"*.news"}
    ))
    public  void  listenTopicQueue2(String msg) throws InterruptedException {
        System.out.println("消费者2--topic接收消息--news：" + msg);
        Thread.sleep(20);
    }

@RequestMapping("/headerQueue")
    public String testHeaderQueue() throws InterruptedException {
        String exchange = "header.exchange";
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setHeader("query","query1");
        //交换机，路由名，信息
        Message message = new Message("info".getBytes(),messageProperties);
        rabbitTemplate.convertAndSend(exchange,"",message);

        return "routeQueue success";
    }

@RabbitListener(queues = "helloworld")
public  void  listenWorkQueue2(Map<String,String> map) throws InterruptedException {
    System.out.println("name：" + map.get("name"));
    System.out.println("age：" + map.get("age"));
    Thread.sleep(50);
}

@Configuration
public class DeathConfig {

    @Bean("normal.exchange")
    public FanoutExchange normalExchange(){
        return new FanoutExchange("normal");
    }

    @Bean("normal.queue")
    public Queue normalQueue(){
        Map<String,Object> args = new HashMap<>();
        //超时
        args.put("x-message-ttl",3000);
        //最大长度
        args.put("x-max-length",5);
        //死信交换机(用死信交换机的名字--rabbitmq当中的名字)
        args.put("x-dead-letter-exchange","death");
        //设置死信队列的路由key
        //args.put("x-dead-letter-routing-key","tec");
        return new Queue("normal",true,false,false,args);
    }

    @Bean
    public Binding normalBinding(@Qualifier("normal.exchange")FanoutExchange fanoutExchange
            ,@Qualifier("normal.queue")Queue queue){
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }


    //死信相关
    @Bean("death.exchange")
    public FanoutExchange deathExchange(){
        return new FanoutExchange("death");
    }

    @Bean("death.queue")
    public Queue DeathQueue(){
        return new Queue("death");
    }

    @Bean
    public Binding deathBinding(@Qualifier("death.exchange")FanoutExchange fanoutExchange
            ,@Qualifier("death.queue")Queue queue){
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }

//定义一个特殊的属性对象
    RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean b, String s) {
            System.out.println("消息发送id：" +  correlationData.getId()  + "，发送状态：" +  b );
        }
    };

    @RequestMapping("/cfm")
    public String testCfm(){
        String message = "hello,死信消息";

        rabbitTemplate.setConfirmCallback(confirmCallback);
        CorrelationData correlationData = new CorrelationData();
        System.out.println("发送的id:" + correlationData.getId());
        rabbitTemplate.convertAndSend("normal","",message ,correlationData);

        return "cfg success";
    }

@RabbitListener(queues = "normal")
    public void normalConsumer(Channel channel, Message message) throws IOException {
        System.out.println("正常消费者：" +message);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        System.out.println("deliveryTag: " + deliveryTag);
        //消费者确认消息
        channel.basicAck(deliveryTag ,true);
        //拒绝消息,requeue为false，则变成死信
        //channel.basicReject(deliveryTag,false);
        //channel.basicReject(deliveryTag,true);
        //未确认消息重新入队
        //channel.basicRecover(true);
    }