spring整合rabbitmq一点小总结_

作者：我家小花儿 | 2024-02-25 06:16:40

赞

踩

1.第一种方式从配置文件中指定交换机和队列以及交换机和队列的绑定关系。
配置文件spring-rabbitmq.xml

<description>rabbitmq 连接服务配置</description>
<!-- 连接配置 -->
<rabbit:connection-factory id="connectionFactory" host="${mq.host}" username="${mq.username}" password="${mq.password}" port="${mq.port}"  virtual-host="${mq.vhost}"/>
<rabbit:admin connection-factory="connectionFactory"/>
<!-- spring template声明-->
<rabbit:template exchange="test-mq-exchange" id="amqpTemplate"  connection-factory="connectionFactory" />
&lt;!&ndash;message-converter="jsonMessageConverter"&ndash;&gt;
<!-- 消息对象json转换类 -->
1
2
3
4
5
6
7
8

<rabbit:direct-exchange name=“test-mq-exchange” durable=“true” auto-delete=“false” id=“test-mq-exchange”>
rabbit:bindings
<rabbit:binding queue=“test_queue_key” key=“test_queue_key”/>
</rabbit:bindings>
</rabbit:direct-exchange>
<rabbit:queue id=“test_queue_key” name=“test_queue_key” durable=“true” auto-delete=“false” exclusive=“false” ></rabbit:queue>

<rabbit:listener-container connection-factory=“connectionFactory” >
<rabbit:listener ref=“amqpConsumer” queue-names=“test_queue_key” />
</rabbit:listener-container>

------------rabbitmq.properties配置文件
mq.host=192.168.75.135
mq.username=amqpadmin
mq.password=123456
mq.port=5672
mq.vhost=/aaa

-----------------生产者发布消息代码
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class AmqpProvider {
@Autowired
private AmqpTemplate amqpTemplate;
public void sendMessage(String routingkey,Object object){

    amqpTemplate.convertAndSend(routingkey,object);
}
1
2

}

消费者消费消息的代码--------
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.stereotype.Component;

public class AmqpConsumer implements MessageListener {

@Override
public void onMessage(Message message) {
    System.out.println(new String(message.getBody())+"哈哈-------asdf--------------");
    MessageProperties messageProperties = message.getMessageProperties();
    System.out.println(message.toString());
}
1
2
3
4
5
6

}

test类测试代码----------------
import com.atcrowdfunding.service.AmqpConsumer;
import com.atcrowdfunding.service.AmqpProvider;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.util.HashMap;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {“classpath:applicationContext.xml”})
public class RabbitmqTest {

@Autowired
private AmqpProvider amqpProvider;
@Test
public void testSend() throws JsonProcessingException {

	String hh = "hsHsdhfsd好合好散--888999-";
  amqpProvider.sendMessage("test_queue_key",hh);

}
1
2
3
4
5
6
7
8
9

}

2.第二种方式代码中指定交换机和队列以及交换机和队列的绑定关系。
配置文件spring-rabbitmq.xml

<description>rabbitmq 连接服务配置</description>
<!-- 连接配置 -->
<rabbit:connection-factory id="connectionFactory" host="${mq.host}" username="${mq.username}" password="${mq.password}" port="${mq.port}"  virtual-host="${mq.vhost}"/>
<rabbit:admin connection-factory="connectionFactory"/>
<!--<bean id="rabbitAdmin" class="org.springframework.amqp.rabbit.core.RabbitAdmin">
    <constructor-arg name="connectionFactory" ref="connectionFactory"></constructor-arg>
</bean>-->

<!-- spring template声明-->
<rabbit:template exchange="admin.exchange.topic" id="amqpTemplate"  connection-factory="connectionFactory" /><!--message-converter="jsonMessageConverter"-->

<!-- 消息对象json转换类 -->
1
2
3
4
5
6
7
8
9
10
11
12

<rabbit:listener-container connection-factory="connectionFactory" >
    <rabbit:listener  ref="amqpConsumer"  queue-names="admin.queue.topic"   />

</rabbit:listener-container>
1
2
3
4

------------rabbitmq.properties配置文件
mq.host=192.168.75.135
mq.username=amqpadmin
mq.password=123456
mq.port=5672
mq.vhost=/aaa

-----------------生产者发布消息代码
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class AmqpProvider {
@Autowired
private AmqpTemplate amqpTemplate;
public void sendMessage(String routingkey,Object object){

    amqpTemplate.convertAndSend(routingkey,object);
}
1
2

}

消费者消费消息的代码--------
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.stereotype.Component;

public class AmqpConsumer implements MessageListener {

@Override
public void onMessage(Message message) {
    System.out.println(new String(message.getBody())+"哈哈-------asdf--------------");
    MessageProperties messageProperties = message.getMessageProperties();
    System.out.println(message.toString());
}
1
2
3
4
5
6

}

test类测试代码--------------
import com.atcrowdfunding.service.AmqpConsumer;
import com.atcrowdfunding.service.AmqpProvider;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.util.HashMap;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {“classpath:applicationContext.xml”})
public class RabbitmqTest {

@Autowired
private AmqpProvider amqpProvider;

@Autowired
private RabbitAdmin rabbitAdmin;

@Test
public void testSend() throws JsonProcessingException {

    //第一种绑定方式
    rabbitAdmin.declareExchange(new TopicExchange("admin.exchange.topic",true,false,null));
    rabbitAdmin.declareQueue(new Queue("admin.queue.topic",true,false,false,null));
    rabbitAdmin.declareBinding(new Binding("admin.queue.topic",Binding.DestinationType.QUEUE,"admin.exchange.topic",
            "admin.topic.#",null));

    //第二种绑定方式
    rabbitAdmin.declareExchange(new DirectExchange("admin.exchange.direct",true,false,null));
    rabbitAdmin.declareQueue(new Queue("admin.queue.direct",true,false,false,null));
    rabbitAdmin.declareBinding(BindingBuilder.bind(
            new Queue("admin.queue.direct",true,false,false,null))
            .to(new DirectExchange("admin.exchange.direct",true,false,null)).with("admin.direct.#"));


    String hh = "hsHsdhfsd好合好散--888999-";
    amqpProvider.sendMessage("admin.topic.insert",hh);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

}

3.第三种方式从rabbitmq管理后台指定交换机和队列以及交换机和队列的绑定关系。