阿里云AMQP Spring集成

概述

本文主要演示如何使用Spring集成开发阿里云的AMQP。

主要配置及步骤

1、pom.xml

    <dependencies>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>1.6.1.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>com.alibaba.mq-amqp</groupId>
            <artifactId>mq-amqp-client</artifactId>
            <version>1.0.3</version>
        </dependency>
    </dependencies>

        

          
        
        
        

          

          AI 代码解读
        
      
      

2、spring-rabbitmq.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd  http://www.springframework.org/schema/context
 http://www.springframework.org/schema/context/spring-context-2.5.xsd  http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

    <!--组件扫描，需要添加pom依赖 spring-context  注意根据自己的包名称配置 -->
    <context:component-scan  base-package="com.alibaba.taro" />

    <bean id="userUtils" class="com.alibaba.mq.amqp.utils.UserUtils"/>

    <!--配置连接-->
    <!--参数获取参考链接：https://yq.aliyun.com/articles/693979?spm=a2c4e.11155435.0.0.4a6126a2PDBrRl -->
    <rabbit:connection-factory id="connectionFactory" host="18****7816617278.mq-amqp.cn-hangzhou-a.aliyuncs.com" port="5672"
                               username="#{userUtils.getUserName('********',18****7816617278L)}" password="#{userUtils.getPassord('********')}"  virtual-host="yutaoamqptest" requested-heartbeat="60" />

    <!--配置RabbitTemplate-->
    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"
                     exchange="demo-test-exchange1" routing-key="foo.bar"/>

    <!--配置RabbitAdmin-->
    <rabbit:admin connection-factory="connectionFactory" />

    <!--配置队列名-->
    <rabbit:queue name="demo-test-queue" />

    <!--配置topic类型exchange-->
    <rabbit:topic-exchange name="demo-test-exchange1">
        <rabbit:bindings>
            <rabbit:binding queue="demo-test-queue" pattern="foo.*" />
        </rabbit:bindings>
    </rabbit:topic-exchange>

    <!--&lt;!&ndash;&lt;!&ndash;配置监听&ndash;&gt;&ndash;&gt;   默认或不设置均为自动确认-->
    <!--<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto">-->
        <!--<rabbit:listener ref="foo" method="onMessage" queue-names="demo-test-queue" />-->
    <!--</rabbit:listener-container>-->

    <!--&lt;!&ndash;配置监听&ndash;&gt;   手动确认-->
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">
        <rabbit:listener ref="receiveConfirmTestListener" method="onMessage" queue-names="demo-test-queue" />
    </rabbit:listener-container>

</beans>

        

          
        
        
        

          

          AI 代码解读
        
      
      

3、自动确认模式消息监听类

package com.alibaba.taro;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.stereotype.Service;

// 自动确认模式的消息消费
@Service
public class Foo implements MessageListener {
    public void onMessage(Message message) {
        try {
            String body=new String(message.getBody(),"UTF-8");
            System.out.println(body);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

        

          
        
        
        

          

          AI 代码解读
        
      
      

4、手动确认模式监听类

package com.alibaba.taro;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.stereotype.Service;

// 带有确认模式的消息消费
@Service("receiveConfirmTestListener")
public class ReceiveConfirmTestListener implements ChannelAwareMessageListener {
    public void onMessage(Message message, Channel channel) throws Exception {
        try{
            System.out.println("consumer--:"+message.getMessageProperties()+":"+new String(message.getBody()));
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }catch(Exception e){
            e.printStackTrace();//TODO 业务处理
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
        }
    }
}

        

          
        
        
        

          

          AI 代码解读
        
      
      

5、main方法

package com.alibaba.taro;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class SpringMain {

    public static void main(final String... args) throws Exception {

        AbstractApplicationContext ctx =
                new ClassPathXmlApplicationContext("spring-rabbitmq.xml");
        RabbitTemplate template = ctx.getBean(RabbitTemplate.class);

        int i = 0;
        while(i<100){
            Thread.sleep(100);
            template.convertAndSend("Hello, world!" + i);
            System.out.println("i: " + i);
            i++;
        }
        //销毁对象
        ctx.destroy();
    }
}

        

          
        
        
        

          

          AI 代码解读
        
      
      

6、项目结构目录

7、测试运行效果

参考链接

AliwareMQ/amqp-demos

RabbitMQ(四)消息确认(发送确认,接收确认)