public class Send {
    @Test
    public void testSend() throws IOException, TimeoutException { 
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机
        connectionFactory.setHost("10.15.0.9");
        //端口号,web管理界面的是15672，而Java链接通信的是5672
        connectionFactory.setPort(5672);
        //设置访问虚拟主机的用户名和密码
        connectionFactory.setUsername("ems");
        connectionFactory.setPassword("123");
        //设置连接那个虚拟主机
        connectionFactory.setVirtualHost("/ems");
        //获取连接对象
        Connection connection = connectionFactory.newConnection();
        //获取连接中的通道
        Channel channel = connection.createChannel();

        //通道绑定对应消息队列  
        //参数1: 队列名称，如果队列不存在会自动创建  
        //参数2: 是否持久化  
        //参数3 exclusive:是否独占队列 
        //参数4 autoDelete:是否在消费完成后自动删除队列
        //参数5:额外附加参数
        channel.queueDeclare("hello",true,false,false,null);
        //发布消息
        //参数1：交换机名称，这里是模型1没有交换机，为空
        //参数2：路由key,没有绑定时默认取队列名作为路由key
        //参数3：传递消息额外设置
        //参数4：消息的具体内容
        channel.basicPublish("","hello", null,"hello rabbitmq".getBytes());
        channel.close();
        connection.close();
}

//受限于Junit版本的问题，会出现消息消费后在回调函数打印之前主线程关闭，从而并未打印信息的情况，这里我们是使用main方法进行测试
public static void main(String[] args) throws IOException, TimeoutException {
    //创建连接工厂
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("10.15.0.9");
    connectionFactory.setPort(5672);
    connectionFactory.setUsername("ems");
    connectionFactory.setPassword("123");
    connectionFactory.setVirtualHost("/ems");
    //创建连接对象
    Connection connection = connectionFactory.newConnection();
    //创建通道
    Channel channel = connection.createChannel();
    //通道绑定消息队列
    channel.queueDeclare("hello", true, false, false, null);
    //消费消息
    //参数1：消费哪个队列的消息，队列名称
    //参数2：开始消息的自动确认机制
    //参数3：消费时的回调接口
    channel.basicConsume("hello",true,new DefaultConsumer(channel){
        @Override
        //最后一个参数：消息队列中取出的消息
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            System.out.println(new String(body));
        }
    });
    //一般来说我们的消费者要一直监听消息队列，所以这里并不需要关闭
    //channel.close();
    //connection.close();
}

/**
* 抽取出重复代码到工具类中
*/
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class RabbitMQUtils {
    private static ConnectionFactory connectionFactory = null;
    static {
        //重量级资源 类加载时执行并且只执行一次
        connectionFactory =  new ConnectionFactory();
        connectionFactory.setHost("192.168.11.128");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/ems");
        connectionFactory.setUsername("ems");
        connectionFactory.setPassword("123");
    }

    //获取连接的方法
    public static Connection getConnection() {
        try{
            return connectionFactory.newConnection();
        }catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }

    //关闭连接的方法
    public static void closeConnectionAndChannel(Channel channel,Connection connection) {
        try {
            if(channel != null) {
                channel.close();
            }
            if(connection != null) {
                connection.close();
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
    }
}

/**
* 生产者中替换成工具类
*/
import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Send {
    @Test
    public void testSend() throws IOException, TimeoutException {
//        ConnectionFactory connectionFactory = new ConnectionFactory();
//        connectionFactory.setHost("192.168.11.128");
//        connectionFactory.setPort(5672);
//        connectionFactory.setUsername("ems");
//        connectionFactory.setPassword("123");
//        connectionFactory.setVirtualHost("/ems");
//        Connection connection = connectionFactory.newConnection();
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("hello",true,false,false,null);
        channel.basicPublish("","hello",null,"hello rabbitmq".getBytes());
//        channel.close();
//        connection.close();
        RabbitMQUtils.closeConnectionAndChannel(channel,connection);
    }
}

/**
* 消费者中替换成工具类
*/
import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

//消费者
public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
//        ConnectionFactory connectionFactory = new ConnectionFactory();
//        connectionFactory.setHost("192.168.11.128");
//        connectionFactory.setPort(5672);
//        connectionFactory.setVirtualHost("/ems");
//        connectionFactory.setUsername("ems");
//        connectionFactory.setPassword("123");

//        Connection connection = connectionFactory.newConnection();
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("hello",true,false,false,null);
        channel.basicConsume("hello",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(new String(body));
            }
        });
    }
}

  channel.queueDeclare("hello",true,false,false,null);
  '参数1':用来声明通道对应的队列，当队列不存在时会自动创建
  '参数2':用来指定是否持久化队列,如果为false,队列不会持久化，重启rabbitmq后队列会消失，注意这里只是队列持久化，队列中的消息需要持久化的话需要通过下面语句进行设置
  '参数3':用来指定是否独占队列，设置为true,只能被一个通道绑定，其他通道在绑定时会报错
  '参数4':用来指定是否自动删除队列，如果设置为true，当队列中的消息被消费完并且队列与消费者完全断开后，队列会自动删除，注意如果消息消费完，消费者还与队列保持连接的情况下是不会删除的，比如这里我们的消费者代码中并未关闭连接，所以不会消息，我们将消费者的后台关闭后队列会消失
  '参数5':对队列的额外配置
      
  //消息进行持久化的设置，在发送消息时进行设置
  channel.basicPublish("","hello", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello rabbitmq".getBytes());
 参数1：交换机名称，可以为空
 参数2：路由key,没有绑定时默认取队列名作为路由key
 参数3：传递消息额外设置，通过这里设置消息进行持久化，这样未消费的消息在重启rabbitmq后也会存在
 参数4：消息的具体内容
     
 需要注意的是，消息消费者要和生产者的队列保持一直，
 比如之前生产者的队列设置了持久化队列为true，这里第二个参数也要设置为true
 之前生产者设置了自动删除为true，那这里第四个参数也要设置为true
 channel.queueDeclare("hello",true,false,false,null);    

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;

public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        //生产20条消息
        for(int i=0;i<20;i++) {
            channel.basicPublish("","work",null,(i + "-hello work").getBytes());
        }
        RabbitMQUtils.closeConnectionAndChannel(channel,connection);
    }
}

mport com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        channel.basicConsume("work",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1：" + new String(body));
            }
        });
    }
}

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        channel.basicConsume("work",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2：" + new String(body));
            }
        });
    }
}

//第二个参数true代表自动确认
channel.basicConsume("hello",true,new DefaultConsumer(channel){……});

一旦开启了自动确认，队列中的消息会一次性的发给消费者，比如一次发送了5条消息，消费者拿到后再消费了3条后宕机了，那么剩下的2条则会丢失掉，所以也会存在消息丢失的可能性

/**
* 生产者代码不变
*
*/
public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        //生产20条消息
        for(int i=0;i<20;i++) {
            channel.basicPublish("","work",null,(i + "-hello work").getBytes());
        }
        RabbitMQUtils.closeConnectionAndChannel(channel,connection);
    }
}



/**
* 消费者1
*channel.basicQos(1);	每次只发一条
*channel.basicConsume("work",false,new DefaultConsumer(channel){})	关闭自动确认
*channel.basicAck(envelope.getDeliveryTag(),false);	手动确认
*/
public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        //告诉消息队列不要全部发送，每次只能消费一个消息，剩余的仍在消息队列中保存，避免消息丢失
        channel.basicQos(1);
//        channel.basicConsume("work",true,new DefaultConsumer(channel){
        //关闭自动分配
        channel.basicConsume("work",false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //为了体现出消费者1和2处理效率的不同，这里让消费者1每次睡1秒
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者1：" + new String(body));
                //手动确认消息
                //参数1：消息标识，用于确认队列中的具体那个消息  参数2：是否开启多个消息同时确认
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        });
    }
}

/**
* 消费者2 和消费者1一样的设置
*channel.basicQos(1);	每次只发一条
*channel.basicConsume("work",false,new DefaultConsumer(channel){})	关闭自动确认
*channel.basicAck(envelope.getDeliveryTag(),false);	手动确认
*/
public class Consumer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        //每次只消费1个消息
        channel.basicQos(1);
//        channel.basicConsume("work",true,new DefaultConsumer(channel){
        //关闭消息自动确认
        channel.basicConsume("work",false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2：" + new String(body));
                //开启自动确认
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        });
    }
}


通过执行可以看到消费者1因为消费的慢只消费了1条消息，而消费者2消费了19条

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;

public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //声明交换机
        //参数1:交换机的名字，如果没有会自动创建  参数2：交换机的类型,对于广播类型，这边必须是 “fanout”
        channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT);
        //channel.exchangeDeclare("logs",“fanout");这种写法也可以
        //发布消息
        channel.basicPublish("logs","",null,"hello fanout message".getBytes());
        RabbitMQUtils.closeConnectionAndChannel(channel,connection);
    }
}

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //通道绑定交换机
        channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT);
        //生成临时队列
        String queue = channel.queueDeclare().getQueue();
        //绑定交换机和队列
        channel.queueBind(queue,"logs","");
        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1：" + new String(body));
            }
        });
    }
}

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //通道绑定交换机
        channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT);
        //生成临时队列
        String queue = channel.queueDeclare().getQueue();
        //绑定交换机和队列
        channel.queueBind(queue,"logs","");
        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2：" + new String(body));
            }
        });
    }
}

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Consumer3 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //通道绑定交换机
        channel.exchangeDeclare("logs", BuiltinExchangeType.FANOUT);
        //生成临时队列
        String queue = channel.queueDeclare().getQueue();
        //绑定交换机和队列
        channel.queueBind(queue,"logs","");
        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者3：" + new String(body));
            }
        });
    }
}

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;

public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("logs_direct", BuiltinExchangeType.DIRECT);
       	//routingKey
        String key = "info";
        channel.basicPublish("logs_direct",key,null,("指定的route key 为"+key+" 的消息").getBytes());
        RabbitMQUtils.closeConnectionAndChannel(channel,connection);
    }
}

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("logs_direct", BuiltinExchangeType.DIRECT);
        String queue = channel.queueDeclare().getQueue();
        //通过指定routingKey为info，只接收对应routingKey相同的信息
        channel.queueBind(queue,"logs_direct","info");
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1："+new String(body));
            }
        });
    }
}

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("logs_direct", BuiltinExchangeType.DIRECT);
        String queue = channel.queueDeclare().getQueue();
        //可以绑定多个，这里可以接收routingKey为error、info、warn的信息
        channel.queueBind(queue,"logs_direct","error");
        channel.queueBind(queue,"logs_direct","info");
        channel.queueBind(queue,"logs_direct","warn");
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2："+ new String(body));
            }
        });
    }
}

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;

public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("topics", BuiltinExchangeType.TOPIC);
        String routekey = "user.save";
        channel.basicPublish("topics",routekey,null,("这是路由中的动态订阅模型,route key: ["+routekey+"]").getBytes());
        RabbitMQUtils.closeConnectionAndChannel(channel,connection);
    }
}

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("topics", BuiltinExchangeType.TOPIC);
        String queue = channel.queueDeclare().getQueue();
        channel.queueBind(queue,"topics","user.*");

        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1: "+new String(body));
            }
        });
    }
}

import com.mg.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("topics", BuiltinExchangeType.TOPIC);
        String queue = channel.queueDeclare().getQueue();
        channel.queueBind(queue,"topics","user.#");

        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2: "+new String(body));
            }
        });
    }
}