RabbitMQ部分API详解(含demo)

demo配置文件rabbitmq.properties

host=192.168.221.138
port=5672
username=guest
password=guest

读取配置文件内容

public class RabbitConfig {
    public static String host ;
    public static int port ;
    public static String username;
    public static String password ;

    static{
        try {
            ClassLoader classLoader=RabbitConfig.class.getClassLoader();
            InputStream is=classLoader.getResourceAsStream("rabbitmq.properties");
            Properties properties=new Properties();
            properties.load(is);
            host = properties.getProperty("host");
            port = Integer.parseInt(properties.getProperty("port"));
            username =properties.getProperty("username");
            password =properties.getProperty("password");
        }catch (IOException e) {
            e.printStackTrace();
        }
    }
}

创建连接工厂

public class RabbitUtil {
    public static Connection GetRabbitConnection() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitConfig.host);
        factory.setPort(RabbitConfig.port);
        factory.setUsername(RabbitConfig.username);
        factory.setPassword(RabbitConfig.password);
        Connection connection = null;
        try {
            connection = factory.newConnection();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return connection;
    }
}

创建provider

public class FanoutProvider {

    public static void main(String[] args) throws Exception {
        // 创建一个连接
        Connection connection = RabbitUtil.GetRabbitConnection();
        // 创建通道
        Channel channel = connection.createChannel();
        //声明exchange
        //exchange:exchange名称
        //type:exchange类型
        //durable:exchange是否持久化(不代表消息持久化)
        //autoDelete:已经没有消费者了,服务器是否可以删除该Exchange
        //exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,Map<String, Object> arguments)
        channel.exchangeDeclare("zx_fanout", BuiltinExchangeType.FANOUT, true,false,null);

        //声明queue
        //queue:queue名称
        //durable:queue是否持久化
        //exclusive:是否为当前连接的专用队列，在连接断开后，会自动删除该队列
        //autodelete：当没有任何消费者使用时，自动删除该队列
        //queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments)
        channel.queueDeclare("zx_queue_f1", true, false, false, null);
        channel.queueDeclare("zx_queue_f2", true, false, false, null);
        channel.queueDeclare("zx_queue_f3", true, false, false, null);

        //queue:queue名称
        //exchange:exchange名称
        //routingKey:路由键;用来绑定queue和exchange
        //queueBind(String queue, String exchange, String routingKey)
        channel.queueBind("zx_queue_f1","zx_fanout","");
        channel.queueBind("zx_queue_f2","zx_fanout","");
        channel.queueBind("zx_queue_f3","zx_fanout","");

        //开启confirm机制
        // (即rabbitmq-server收到生产端的消息,会给生产端发送一个确认,如果没法送确认,生产端重新发送消息到server中)
        channel.confirmSelect();

        for (int i = 1 ; i <= 100 ;i++){
            String message = "fanout:广播的第 "+ i +" 条消息";

            //exchange
            //routingKey：路由键
            //mandatory：true=如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue，那么会调用basic.return方法将消息返还给生产者。false=出现上述情形broker会直接将消息扔掉
            //immediate：true=如果exchange在将消息route到queue(s)时发现对应的queue上没有消费者，那么这条消息不会放入队列中。false=当与消息routeKey关联的所有queue(一个或多个)都没有消费者时，该消息会通过basic.return方法返还给生产者。
            //BasicProperties：消息的基本属性，例如路由头等(MessageProperties.PERSISTENT_TEXT_PLAIN:消息的持久化)
            //basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)
            channel.basicPublish("zx_fanout","",false,false,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("utf-8"));
            System.out.println(message);
        }

        // 关闭连接
        channel.close();
        connection.close();
    }
}

创建consumer

public class Consumer1 {
    public static void main(String[] args) throws Exception {
        // 创建一个连接
        Connection conn = RabbitUtil.GetRabbitConnection();
        // 创建通道
        Channel channel = conn.createChannel();
        //声明exchange
        //exchange:exchange名称
        //type:exchange类型
        //durable:exchange是否持久化(不代表消息持久化)
        //autoDelete:已经没有消费者了,服务器是否可以删除该Exchange
        //exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,Map<String, Object> arguments)
        channel.exchangeDeclare("zx_fanout", BuiltinExchangeType.FANOUT,true,false,null);

        //声明queue
        //queue:queue名称
        //durable:queue是否持久化
        //exclusive:是否为当前连接的专用队列，在连接断开后，会自动删除该队列
        //autodelete：当没有任何消费者使用时，自动删除该队列
        //queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments)
        channel.queueDeclare("zx_queue_f1", true, false, false, null);

        //queue:queue名称
        //exchange:exchange名称
        //routingKey:路由键;用来绑定queue和exchange
        //queueBind(String queue, String exchange, String routingKey)
        channel.queueBind("zx_queue_f1","zx_fanout","");
        
        //uack的消息最多有几条
        channel.basicQos(200);

        //消息消费
        //queue:绑定队列名称
        //autoAck：是否自动ack，如果不自动ack，需要使用channel.ack、channel.nack、channel.basicReject 进行消息应答
        //basicConsume(String queue, boolean autoAck, Consumer callback)
        channel.basicConsume("zx_queue_f1",false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    String message = new String(body,"utf-8");
                    System.out.println(Consumer1.class.getSimpleName()+"--接收消息--"+message);
                } catch (UnsupportedEncodingException e) {

                    //deliveryTag:该消息的唯一ID
                    //multiple：是否批量. true:将一次性拒绝所有小于deliveryTag的消息。
                    //requeue：被拒绝的是否重新入队列
                    //channel.basicNack(long deliveryTag, boolean multiple, boolean requeue);
                    channel.basicNack(envelope.getDeliveryTag(),false,true);

                    //deliveryTag:该消息的index
                    //requeue：被拒绝的是否重新入队列
                    //channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
                    //channel.basicNack 与 channel.basicReject 的区别在于basicNack可以拒绝多条消息，而basicReject一次只能拒绝一条消息
                } finally {

                    //basicAck(long deliveryTag, boolean multiple)
                    //deliveryTag:该消息的唯一ID
                    //multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        });
    }
}

demo的github地址

https://github.com/AndeZhaoX/rabbitmq-test.git