ssm整合rabbitmq消息队列的简单使用案例

柔情只为你懂 2022-05-09 11:04 387阅读 1赞

项目：ssm（spring+springmvc+mybatis）

编译工具：eclipse

消息队列：rabbitmq

1.在pom.xml添加rabbitmq依赖：

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

2.编写rabbitmq的配置文件：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:rabbit="http://www.springframework.org/schema/rabbit"
        xsi:schemaLocation="http://www.springframework.org/schema/beans  
         http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  
         http://www.springframework.org/schema/rabbit  
         http://www.springframework.org/schema/rabbit/spring-rabbit-1.2.xsd">
        
        <rabbit:connection-factory id="connectionFactory" 
            username="guest" password="guest" host="127.0.0.1" port="5672" 
            />
        
    
        
        <rabbit:admin id="connectAdmin" connection-factory="connectionFactory" />
    
        
        <rabbit:queue name="queueTest" durable="true" auto-delete="false"
            exclusive="false" declared-by="connectAdmin" />
    
        
        <rabbit:direct-exchange name="exchangeTest"
            durable="true" auto-delete="false" declared-by="connectAdmin">
            <rabbit:bindings>
                <rabbit:binding queue="queueTest" key="queueTestKey"></rabbit:binding>
            </rabbit:bindings>
        </rabbit:direct-exchange>
    
        
        <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"
            exchange="exchangeTest" />
    
        
        <bean id="messageReceiver" class="com.comit.appointment.modules.test.MessageConsumer"></bean>
    
        
        <rabbit:listener-container
            connection-factory="connectionFactory">
            <rabbit:listener queues="queueTest" ref="messageReceiver" />
        </rabbit:listener-container>
    
        
        <rabbit:queue name="queueChris" durable="true"
            auto-delete="false" exclusive="false" declared-by="connectAdmin" />
    
        
        <rabbit:direct-exchange name="exchangeTest"
            durable="true" auto-delete="false" declared-by="connectAdmin">
            <rabbit:bindings>
                <rabbit:binding queue="queueChris" key="queueTestChris"></rabbit:binding>
            </rabbit:bindings>
        </rabbit:direct-exchange>
    
        
        <bean id="receiverChris" class="com.comit.appointment.modules.test.ChrisConsumer"></bean>
    
        
        <rabbit:listener-container
            connection-factory="connectionFactory">
            <rabbit:listener queues="queueChris" ref="receiverChris" />
        </rabbit:listener-container>
    
        
        
        <rabbit:connection-factory id="connectionFactory2"  
            username="guest" password="guest" host="127.0.0.1" port="5672" />
    
        
        <rabbit:admin id="connectAdmin2" connection-factory="connectionFactory2" />
    
        
        <rabbit:queue name="queueShijj" durable="true"
            auto-delete="false" exclusive="false" declared-by="connectAdmin2" />
    
        
        <rabbit:topic-exchange name="exchangeTest2"
            durable="true" auto-delete="false" declared-by="connectAdmin2">
            <rabbit:bindings>
                <rabbit:binding queue="queueShijj" pattern="shijj.#"></rabbit:binding>
            </rabbit:bindings>
        </rabbit:topic-exchange>
    
        
        <rabbit:template id="amqpTemplate2" connection-factory="connectionFactory2"
            exchange="exchangeTest2" />
    
        
        <bean id="recieverShijj" class="com.comit.appointment.modules.test.ShijjConsumer"></bean>
    
        
        <rabbit:listener-container
            connection-factory="connectionFactory2">
            <rabbit:listener queues="queueShijj" ref="recieverShijj" />
        </rabbit:listener-container>
    </beans>

3.在ssm项目的上下文配置文件spring-content.xml中引入rabbitmq的配置文件

spring-content.xml:

4.编写测试代码：

1.消息生产者：

package com.comit.appointment.modules.test.cs;
    
    import java.io.IOException;
    
    import javax.annotation.Resource;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.amqp.core.Message;
    import org.springframework.stereotype.Service;
    
    import com.comit.appointment.modules.sys.entity.Users;
    
    @Service("MessageProducer2")
    public class MessageProducer2 {
    	
    	@Resource(name="amqpTemplate")
    	private AmqpTemplate  amqpTemplate;
    	
    	public void sendMessage(Object message)throws IOException{
    		amqpTemplate.convertAndSend("queueTestKey",message);
    	}
    	
    	
    
    }

注：queueTestKey 是在rabbitMQ.xml绑定到队列的，根据这个key来发送消息到相应的消息队列中。

<rabbit:binding queue="queueTest" key="queueTestKey"></rabbit:binding>

2.消息消费者：获取消息并把消息插入日志记录表中

package com.comit.appointment.modules.test;
    
    
    import java.io.UnsupportedEncodingException;
    import java.text.ParseException;
    import java.text.SimpleDateFormat;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    import org.springframework.beans.factory.annotation.Autowired;
    
    import com.comit.appointment.modules.sys.dao.LogsMapper;
    import com.comit.appointment.modules.sys.entity.Logs;
    
    
    public class MessageConsumer implements MessageListener {
         private Logger logger = LoggerFactory.getLogger(MessageConsumer.class); 
         
         @Autowired
         private LogsMapper logsMapper;
         
        @Override
        public void onMessage(Message message) {
        	  try {
    			String messages = new String(message.getBody(), "UTF-8");
    			String[] str=messages.split(",");
    			Logs log=new Logs();
    			log.setNAME(str[0]);
    			log.setTIME(str[1]);
    			log.setACTION(str[2]);
    			System.out.println("log-->"+log.toString());
    			logsMapper.insertSelective(log);
    		} catch (UnsupportedEncodingException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
             logger.info("consumer receive message------->:{}", message);  
        }
    
    }

在登录时发送已登录的消息：

package com.comit.appointment.modules.sys.controller.impl;
    
    import java.io.IOException;
    import java.util.Date;
    
    import javax.servlet.http.HttpSession;
    
    import org.apache.shiro.session.Session;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestMapping;
    
    import com.comit.appointment.modules.sys.controller.IBaseController;
    import com.comit.appointment.modules.sys.entity.Users;
    import com.comit.appointment.modules.sys.service.IOperaterService;
    import com.comit.appointment.modules.sys.service.IUsersService;
    import com.comit.appointment.modules.test.cs.MessageProducer2;
    
    @Controller
    @RequestMapping("/base")
    public class BaseController implements IBaseController {
    	
    	@Autowired
    	private IUsersService usersService;
    	
    	@Qualifier(value="MessageProducer2")
    	@Autowired
    	private MessageProducer2 messageProducer;
    
    	@RequestMapping("/")
    	public String login() {
    		return "modules/sys/login";
    	}
    	
    	@Override
    	@RequestMapping("/login")
    	public String login(HttpSession session,Users user) {
    		Users user1=usersService.findUser(user);
    		if(user1!=null) {
    			try {
    				messageProducer.sendMessage(user1.getACCOUNT()+","+new Date()+",登录了");
    			} catch (IOException e) {
    				e.printStackTrace();
    			}
    			session.setAttribute("user", user1);
    			return "modules/sys/index";
    		}
    		return "modules/sys/login";
    	}
    	
    	
    
    }

在登录时便发送了一个消息，控制台打印：

![70][] 在消息消费者那里把该消息插入了日志记录表：

![70 1][]

### 注：运行项目时记得打开rabbitmq服务，要不然连接不上会报错。 ###

没有安装rabbitmq的可以看看 [Win10安装RabbitMQ][Win10_RabbitMQ]

### 我的座右铭：不会，我可以学；落后，我可以追赶；跌倒，我可以站起来；我一定行。 ###

[70]: /images/20220509/e382a2912a534f5caf58ba8f22c6f375.png
[70 1]: /images/20220509/c68fbea277f74bf6ac3854de84d0d31d.png
[Win10_RabbitMQ]: https://blog.csdn.net/weixin_39220472/article/details/82934860