zoukankan html css js c++ java

Rabbitmq基本API使用

一、生产者

创建ConnectionFactory工厂（地址、用户名、密码、vhost）
创建Connection
创建信道（Channel）
创建 exchange（指定名称、类型-DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");、是否持久化）
发送消息（指定：exchange、发送的routingKey ，发送到的消息）

基础的生产者：

public class TestProducer {


    public final static String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args)
            throws IOException, TimeoutException {
        /* 创建连接,连接到RabbitMQ*/
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.112.131");
        connectionFactory.setVirtualHost("my_vhost");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        Connection connection = connectionFactory.newConnection();


        /*创建信道*/
        Channel channel = connection.createChannel();
        /*创建交换器*/
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");
        //channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT);


        /*日志消息级别，作为路由键使用*/
        String[] routekeys = {"king","queue","prince"};
        for(int i=0;i<3;i++){
            String routekey = routekeys[i%3];
            String msg = "Hellol,RabbitMq"+(i+1);
            /*发布消息，需要参数：交换器，路由键，其中以日志消息级别为路由键*/
            channel.basicPublish(EXCHANGE_NAME,routekey,null,
                    msg.getBytes());
            System.out.println("Sent "+routekey+":"+msg);
        }
        channel.close();
        connection.close();
    }
}

View Code

二、消费者

创建ConnectionFactory工厂（地址、用户名、密码、vhost）
创建Connection
创建信道（Channel）
声明一个 exchange（指定名称、类型、是否持久化）
创建一个队列（指定：名称，是否持久化，是否独占，是否自动删除，其他参数）
队列、exchange通过routeKey进行绑定
消费者接收消息（队列名称，是否自动ACK）

基本的消费者：

public class TestConsumer {
    public static void main(String[] argv)
            throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.112.131");
        factory.setVirtualHost("my_vhost");
        factory.setUsername("admin");
        factory.setPassword("admin");
        // 打开连接和创建频道，与发送端一样
        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();
        channel.exchangeDeclare(TestProducer.EXCHANGE_NAME,
                "direct");
        /*声明一个队列*/
        String queueName = "focuserror";
        channel.queueDeclare(queueName,false,false,
                false,null);

        /*绑定，将队列和交换器通过路由键进行绑定*/
        String routekey = "king";/*表示只关注error级别的日志消息*/
        channel.queueBind(queueName,TestProducer.EXCHANGE_NAME,routekey);
        System.out.println("waiting for message........");
        /*声明了一个消费者*/
        final Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Received["+envelope.getRoutingKey()
                        +"]"+message);
            }
        };
        /*消费者正式开始在指定队列上消费消息*/
        channel.basicConsume(queueName,true,consumer);
    }
}

View Code

三、消息持久化

exchange 需要持久化
发送消息设置参数为 MessageProperties.PERSISTENT_TEXT_PLAIN
队列需要设置参数为持久化

1、//TODO 创建持久化交换器 durable=true
channel.exchangeDeclare(EXCHANGE_NAME,"direct",true);

2、//TODO 发布持久化的消息(delivery-mode=2)
channel.basicPublish(EXCHANGE_NAME,routekey,
        MessageProperties.PERSISTENT_TEXT_PLAIN,
        msg.getBytes());

3、//TODO 声明一个持久化队列(durable=true)
// autoDelete=true 消费者停止了，则队列会自动删除
//exclusive=true独占队列，只能有一个消费者消费
String queueName = "msgdurable";
channel.queueDeclare(queueName,true,false,
        false,null);

四、如何支持事务（防止投递消息的时候消息丢失-效率特别低，不建议使用，可以使用生产者ACK机制）

启动事务
成功提交
失败则回滚

//TODO
//加入事务
channel.txSelect();
try {
    for(int i=0;i<3;i++){
        String routekey = routekeys[i%3];
        // 发送的消息
        String message = "Hello World_"+(i+1)
                +("_"+System.currentTimeMillis());
        channel.basicPublish(EXCHANGE_NAME, routekey, true,
                null, message.getBytes());
        System.out.println("----------------------------------");
        System.out.println(" Sent Message: [" + routekey +"]:'"
                + message + "'");
        Thread.sleep(200);
    }
    //TODO
    //事务提交
    channel.txCommit();
} catch (IOException e) {
    e.printStackTrace();
    //TODO
    //事务回滚
    channel.txRollback();
} catch (InterruptedException e) {
    e.printStackTrace();
}

View Code

五、消费消息手动ACK，如果异常则使用拒绝的方式，然后异常消息推送到-死信队列

批量ack的时候如果其中有一个消息出现异常，则会导致消息丢失（日志处理的时候可以使用批量）

1 /*消费者正式开始在指定队列上消费消息，第二个参数false为手动应答*/
channel.basicConsume(queueName,false,consumer);

2 收到消息以后，手动应答数据接收成功
channel.basicAck(
        envelope.getDeliveryTag(),false);
3 收到消息，如果处理失败则拒绝消息：DeliveryTag是消息在队列中的标识
channel.basicReject(
        envelope.getDeliveryTag(),false);
4 决绝的参数说明
//TODO Reject方式拒绝(这里第2个参数决定是否重新投递),不要重复投递，因为消息重复投递后处理可能依然异常
//channel.basicReject(envelope.getDeliveryTag(),false);

//TODO Nack方式的拒绝（第2个参数决定是否批量,第3个参数是否重新投递）
channel.basicNack(envelope.getDeliveryTag(), false, true);

View Code

六、创建队列的参数解析：场景，延迟队列，保存带有时效性的订单，一旦订单过期，则信息会转移到死信队列

//TODO /*自动过期队列--参数需要Map传递*/
String queueName = "setQueue";
Map<String, Object> arguments = new HashMap<String, Object>();
arguments.put("x-expires",10*1000);//消息在队列中保存10秒后被删除
//TODO 队列的各种参数
/*加入队列的各种参数*/
// autoDelete=true 消费者停止了，则队列会自动删除
//exclusive=true独占队列，只能有一个消费者消费
channel.queueDeclare(queueName,true,true, false,arguments);

七、发送消息以后带有应答的队列

声明一个回应队列
声明一个回应消息的消费者
声明一个属性对象（指定队列，会唯一的id）
生产者发送消息给消费者（带着回应队列）
消费者接收到消息以后根据对应的信息，给予回应

生产者端：

1、
//TODO 响应QueueName ，消费者将会把要返回的信息发送到该Queue
String responseQueue = channel.queueDeclare().getQueue();
//TODO 消息的唯一id
String msgId = UUID.randomUUID().toString();

2、
/*声明了一个消费者*/
final Consumer consumer = new DefaultConsumer(channel){
    @Override
    public void handleDelivery(String consumerTag,
                               Envelope envelope,
                               AMQP.BasicProperties properties,
                               byte[] body) throws IOException {
        String message = new String(body, "UTF-8");
        System.out.println("Received["+envelope.getRoutingKey()
                +"]"+message);
    }
};
//TODO 消费者应答队列上的消息
channel.basicConsume(responseQueue,true,consumer);
3、
//TODO 设置消息中的应答属性
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .replyTo(responseQueue)
        .messageId(msgId)
        .build();
4、
String msg = "Hello,RabbitMq";
//TODO 发送消息时，把响应相关属性设置进去
channel.basicPublish(EXCHANGE_NAME,"error",
        properties,
        msg.getBytes());

View Code

消费者端：

String message = new String(body, "UTF-8");
System.out.println("Received["+envelope.getRoutingKey()
        +"]"+message);
//TODO 从消息中拿到相关属性（确定要应答的消息ID,）
AMQP.BasicProperties respProp
        = new AMQP.BasicProperties.Builder()
        .replyTo(properties.getReplyTo())
        .correlationId(properties.getMessageId())
        .build();
//TODO 消息消费时，同时需要生作为生产者生产消息（以OK为标识）
channel.basicPublish("", respProp.getReplyTo() ,
        respProp ,
        ("OK,"+message).getBytes("UTF-8"));

八、死信队列 - 下列消息会放到死信队列

消息被否定确认，使用 channel.basicNack 或 channel.basicReject ，并且此时requeue 属性被设置为false。
消息在队列的存活时间超过设置的TTL时间。
消息队列的消息数量已经超过最大队列长度

 测试，消费者被拒绝的时候消息会进到死信队列中：
 
        final Channel channel = connection.createChannel();
        channel.exchangeDeclare(WillMakeDlxConsumer.BUS_EXCHANGE_NAME,
                BuiltinExchangeType.TOPIC);
        //TODO 绑定死信交换器
        /*声明一个队列，并绑定死信交换器*/


        Map<String,Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", WillMakeDlxConsumer.DLX_EXCHANGE_NAME);
//        //TODO 死信路由键，会替换消息原来的路由键
//        args.put("x-dead-letter-routing-key", "deal");


        channel.queueDeclare(WillMakeDlxConsumer.BUS_QUEUE_NAME,false,false,
                false,
                args);


        /*绑定，将队列和交换器通过路由键进行绑定*/
        channel.queueBind(WillMakeDlxConsumer.BUS_QUEUE_NAME,
                WillMakeDlxConsumer.BUS_EXCHANGE_NAME,"#");


        System.out.println("waiting for message........");


        //声明死信队列
        channel.exchangeDeclare(WillMakeDlxConsumer.DLX_EXCHANGE_NAME, "topic",true);
        channel.queueDeclare(WillMakeDlxConsumer.DLX_QUEUE_NAME, true, false, false, null);
        //路由键为 # 代表可以路由到所有消息
        channel.queueBind(WillMakeDlxConsumer.DLX_QUEUE_NAME ,WillMakeDlxConsumer.DLX_EXCHANGE_NAME,  "#");


        /*声明了一个消费者*/
        final Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                //TODO
                //TODO 如果是king的消息确认
                if(envelope.getRoutingKey().equals("king")){
                    System.out.println("Received["
                            +envelope.getRoutingKey()
                            +"]"+message);
                    channel.basicAck(envelope.getDeliveryTag(),
                            false);
                }else{
                    //TODO 如果是其他的消息拒绝（rqueue=false），成为死信消息
                    System.out.println("Will reject["
                            +envelope.getRoutingKey()
                            +"]"+message);
                    channel.basicReject(envelope.getDeliveryTag(),
                            false);
                }
            }
        };
        /*消费者正式开始在指定队列上消费消息*/
        channel.basicConsume(WillMakeDlxConsumer.BUS_QUEUE_NAME ,false,consumer);

View Code

过期消息进入到死信队列：

public static void main(String[] argsv) throws IOException, TimeoutException {


    Connection connection = getConnection();

    // 创建一个信道
    Channel channel = connection.createChannel();
    // 指定转发
    channel.exchangeDeclare(WillMakeDlxConsumer.BUS_EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

    Map<String,Object> args = new HashMap<>();
    args.put("x-dead-letter-exchange", WillMakeDlxConsumer.DLX_EXCHANGE_NAME);
    //设置队列中消息过期时间，过期则进入死信队列
    args.put("x-message-ttl", 10*1000);


    channel.queueDeclare(WillMakeDlxConsumer.BUS_QUEUE_NAME,true,false,
            false,
            args);


    /*绑定，将队列和交换器通过路由键进行绑定*/
    channel.queueBind(WillMakeDlxConsumer.BUS_QUEUE_NAME,
            WillMakeDlxConsumer.BUS_EXCHANGE_NAME,"#");


    System.out.println("waiting for message........");


    //声明死信队列
    channel.exchangeDeclare(WillMakeDlxConsumer.DLX_EXCHANGE_NAME, "topic",true);
    channel.queueDeclare(WillMakeDlxConsumer.DLX_QUEUE_NAME, true, false, false, null);
    //路由键为 # 代表可以路由到所有消息
    channel.queueBind(WillMakeDlxConsumer.DLX_QUEUE_NAME ,WillMakeDlxConsumer.DLX_EXCHANGE_NAME,  "#");


    /*日志消息级别，作为路由键使用*/
    String[] routekeys = {"king","mark","james"};
    for(int i=0;i<10;i++){
        String routekey = routekeys[i%3];
        String msg = "Hellol,RabbitMq"+(i+1);
        /*发布消息，需要参数：交换器，路由键，其中以日志消息级别为路由键*/
        channel.basicPublish(WillMakeDlxConsumer.BUS_EXCHANGE_NAME,routekey,null,
                msg.getBytes());
        System.out.println("Sent "+routekey+":"+msg);
    }
    // 关闭频道和连接
    channel.close();
    connection.close();
}

View Code

九、消费者批量预取消费-每次服务器给消费者推送多少数据进行处理

//TODO 如果是两个消费者(QOS ,批量)则轮询获取数据

//TODO 150条预取(150都取出来 150， 210-150  60  )
channel.basicQos(150,true);
/*消费者正式开始在指定队列上消费消息*/
channel.basicConsume(queueName,false,consumer);
//TODO 自定义消费者批量确认
//BatchAckConsumer batchAckConsumer = new BatchAckConsumer(channel);
//channel.basicConsume(queueName,false,batchAckConsumer);

View Code

十、生产者投递消息确认模式，如果失败了则可以重新投递

同步确认：

// 启用发送者确认模式
channel.confirmSelect();


//所有日志严重性级别
for(int i=0;i<2;i++){
    // 发送的消息
    String message = "Hello World_"+(i+1);
    //参数1：exchange name
    //参数2：routing key
    channel.basicPublish(EXCHANGE_NAME, ROUTE_KEY, true,null, message.getBytes());
    System.out.println(" Sent Message: [" + ROUTE_KEY +"]:'"+ message + "'");
    //TODO
    //确认是否成功(true成功)
    if(channel.waitForConfirms()){
        System.out.println("send success");
    }else{
        //如果失败则，可以重新投递减小消息丢失的几率
        System.out.println("send failure");
    }
}

View Code

异步确认-添加监听器:

//TODO
// 启用发送者确认模式
channel.confirmSelect();
//TODO
// 添加发送者确认监听器
channel.addConfirmListener(new ConfirmListener() {
    //TODO 成功
    public void handleAck(long deliveryTag, boolean multiple)
            throws IOException {
        System.out.println("send_ACK:"+deliveryTag+",multiple:"+multiple);
    }
    //TODO 失败
    public void handleNack(long deliveryTag, boolean multiple)
            throws IOException {
        System.out.println("Erro----send_NACK:"+deliveryTag+",multiple:"+multiple);
    }
});


//TODO
// 添加失败者通知
channel.addReturnListener(new ReturnListener() {
    public void handleReturn(int replyCode, String replyText,
                             String exchange, String routingKey,
                             AMQP.BasicProperties properties,
                             byte[] body)
            throws IOException {
        String message = new String(body);
        System.out.println("RabbitMq路由失败:  "+routingKey+"."+message);
    }
});




String[] routekeys={"king","mark"};
//TODO 6条
for(int i=0;i<20;i++){
    String routekey = routekeys[i%2];
    //String routekey = "king";
    // 发送的消息
    String message = "Hello World_"+(i+1)+("_"+System.currentTimeMillis());
    channel.basicPublish(EXCHANGE_NAME, routekey, true,
            MessageProperties.PERSISTENT_BASIC, message.getBytes());
}

View Code

十一、基本信息-创建链接

public class BasicMq {

    public static final String  MQ_IP = "192.168.112.131";
    public static final String  USER = "admin";
    public static final String  PWD = "admin";
    public static final String  VHOST = "my_vhost";

    /**
     *
     * @return
     */
    public static final Connection getConnection()   {
        /* 创建连接,连接到RabbitMQ*/
        ConnectionFactory connectionFactory = null;
        try {
            connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(BasicMq.MQ_IP);
            connectionFactory.setUsername( BasicMq.USER);
            connectionFactory.setPassword( BasicMq.PWD );
            connectionFactory.setVirtualHost( BasicMq.VHOST );

            return connectionFactory.newConnection();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null ;
    }


    public static final Channel getChannel() throws IOException {
        return getConnection().createChannel();
    }
}

View Code

十二、异常整理

如果生产者声明exchange为 durable=true，那么消费者对应的exchange也必须为durable=true
消费者原来是durable=false，修改后变为durable=true，那么因为服务器上已经有这个队列，但是参数不一致会异常，需要删除服务器上的对应队列

查看全文

相关阅读:
数据库 —— 基于 ORM 模型的 Hibernate 的使用（java）
数据库 —— mySQL 的安装
 数据库 —— 应用程序与数据库的连接
 windows 编程 —— 子窗口类别化（Window Subclassing）
windows 编程 —— 消息与参数（定时器、初始化消息、改变大小）
windows 编程 —— 子窗口与子窗口控件
 windows 编程 —— 消息与参数（滚动条、键盘、鼠标）
windows 编程—— 使用函数笔记
 关于计算机编程语言——编译型和解释型_2
关于计算机编程语言——编译型和解释型【转】

原文地址：https://www.cnblogs.com/lean-blog/p/14150887.html