经过前段时间的学习和铺垫，已经对spring amqp有了大概的了解。俗话说学以致用，今天就利用springAMQP来完成一个日志管理模块。大概的需求是这样的：系统中有很多地方需要记录操作日志，比如登录、退出、查询等，如果将记录日志这个操作掺杂在主要的业务逻辑当中，势必会增加响应的时间，对客户来说是一种不好的体验。所以想到用异步消息队列来进行优化。系统处理完主要业务逻辑之后，将日志的相关实体发布到特定Queue下，然后设置一个监听器，监该Queue的消息并做处理。客户不用等待日志的处理就可直接返回。
大概的业务流程如下图所示。
1.首先建立日志的数据表和实体，数据表起名为t_log。实体如下。主要包含操作者，操作的事件，操作时间等几个主要参数。
package com.xdx.entity;import java.util.Date;public class TLog { private Integer logId; private String operator; private String event; private Date createTime; private Integer isDel; public TLog(String operator, String event) { this.operator = operator; this.event = event; } public TLog() { } public Integer getLogId() { return logId; } public void setLogId(Integer logId) { this.logId = logId; } public String getOperator() { return operator; } public void setOperator(String operator) { this.operator = operator == null ? null : operator.trim(); } public String getEvent() { return event; } public void setEvent(String event) { this.event = event == null ? null : event.trim(); } public Date getCreateTime() { return createTime; } public void setCreateTime(Date createTime) { this.createTime = createTime; } public Integer getIsDel() { return isDel; } public void setIsDel(Integer isDel) { this.isDel = isDel; }}
2.编写保存日志的方法，很简单，就是一个数据库的save过程。
package com.xdx.service;import javax.annotation.Resource;import org.springframework.stereotype.Service;import com.xdx.dao.BaseDao;import com.xdx.entity.TLog;@Servicepublic class LogService { @Resource(name = "baseDao") private BaseDao<TLog, Integer> baseDao; public Integer saveLog(TLog log) { Integer result = baseDao.addT("TLogMapper.insertSelective", log); return result; }}
其中的TLogMapper.insertSelective代码如下：
<insert id="insertSelective" parameterType="com.xdx.entity.TLog" > insert into t_log <trim prefix="(" suffix=")" suffixOverrides="," > <if test="logId != null" > log_id, </if> <if test="operator != null" > operator, </if> <if test="event != null" > event, </if> <if test="createTime != null" > create_time, </if> <if test="isDel != null" > is_del, </if> </trim> <trim prefix="values (" suffix=")" suffixOverrides="," > <if test="logId != null" > #{logId,jdbcType=INTEGER}, </if> <if test="operator != null" > #{operator,jdbcType=VARCHAR}, </if> <if test="event != null" > #{event,jdbcType=VARCHAR}, </if> <if test="createTime != null" > #{createTime,jdbcType=TIMESTAMP}, </if> <if test="isDel != null" > #{isDel,jdbcType=INTEGER}, </if> </trim> </insert>
3.接下来就跟我们的spring amqp有关了，首先要在pom.xml中引入相关的jar包。
<!-- spring-rabbitMQ --> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>2.0.1.RELEASE</version> </dependency> <!-- spring -amqp --> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-amqp</artifactId> <version>2.0.1.RELEASE</version> </dependency>
4.编写主配置文件。在项目新建一个com.xdx.spring_rabbit包。关于rabbit的所有代码都写在这边。
编写一个抽象的rabbit的主配置文件，之所以这样做是为了以后扩展方便，让不同的异步消息队列的业务可以继承并扩展它。如下所示。
主配置的文件主要是配置了连接Rabbit服务的基本信息，并且指定了消息转换器是json转换器。
package com.xdx.spring_rabbit;import org.springframework.amqp.core.AmqpAdmin;import org.springframework.amqp.core.TopicExchange;import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;import org.springframework.amqp.rabbit.connection.ConnectionFactory;import org.springframework.amqp.rabbit.core.RabbitAdmin;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;import org.springframework.amqp.support.converter.MessageConverter;import org.springframework.beans.factory.annotation.Value;import org.springframework.context.annotation.Bean;/** * 抽象类，rabbitMQ的主配置类 * * @author xdx * */public abstract class AbstractRabbitConfiguration { @Value("${amqp.port:5672}") private int port = 5672; protected abstract void configureRabbitTemplate(RabbitTemplate template); /** * 由于connectionFactory会与项目中的redis的connectionFactory命名冲突， * 所以这边改名为rabbit_connectionFactory * * @return */ @Bean public ConnectionFactory rabbit_connectionFactory() { CachingConnectionFactory connectionFactory = new CachingConnectionFactory( "192.168.1.195"); connectionFactory.setUsername("xdx"); connectionFactory.setPassword("xxxx"); connectionFactory.setPort(port); return connectionFactory; } @Bean public RabbitTemplate rabbitTemplate() { RabbitTemplate template = new RabbitTemplate(rabbit_connectionFactory()); template.setMessageConverter(jsonMessageConverter()); configureRabbitTemplate(template); return template; } @Bean public MessageConverter jsonMessageConverter() { return new Jackson2JsonMessageConverter(); } @Bean public AmqpAdmin amqpAdmin() { RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbit_connectionFactory()); return rabbitAdmin; }}
5.编写我们这个日志项目需要用到的配置文件，继承上述的抽象类，在该配置文件中，我们具体指定Exchange，RouteKey，Queue,Binding以及监听器这些要素。
package com.xdx.spring_rabbit;import org.springframework.amqp.core.AcknowledgeMode;import org.springframework.amqp.core.Binding;import org.springframework.amqp.core.BindingBuilder;import org.springframework.amqp.core.Queue;import org.springframework.amqp.core.TopicExchange;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;/** * 日志管理的Rabbit配置类的具体实现类 * * @author xdx * */@Configurationpublic class LogRabbitConfiguration extends AbstractRabbitConfiguration { protected static String LOG_EXCHANGE_NAME = "warrior.exchange.log";// topic // exchange的名称 protected static String LOG_QUEUE_NAME = "warrior.queue.log";// 接收消息的queue protected static String LOG_ROUTING_KEY = LOG_QUEUE_NAME; @Autowired private LogRabbitRecHandler logRabbitRecHandler;// 监听器的委托类，委托其处理接收到的消息 /** * 设置Exchange为LOG_EXCHANGE_NAME，RoutingKey为LOG_ROUTING_KEY，这样将信息发送到 * Exchange为LOG_EXCHANGE_NAME，RouteKey为LOG_ROUTING_KEY的通道中 */ @Override protected void configureRabbitTemplate(RabbitTemplate template) { System.err.println("创建一个RabbitTemplate，名字是 " + template); template.setExchange(LOG_EXCHANGE_NAME); template.setRoutingKey(LOG_ROUTING_KEY); } /** * 用于接收日志消息的Queue,默认绑定自己的名称 * * @return */ @Bean public Queue logQueue() { return new Queue(LOG_QUEUE_NAME); } /** * 定义一个topExchange * * @return */ @Bean public TopicExchange logExchange() { return new TopicExchange(LOG_EXCHANGE_NAME); } /** * 定义一个绑定日志接收的Queue的binding * * @return */ @Bean public Binding logQueueBinding() { return BindingBuilder.bind(logQueue()).to(logExchange()) .with(LOG_ROUTING_KEY); } /** * 这个bean为监听适配器，用于日志消息，并交由logRabbitRecHandler处理 * * @return */ @Bean public MessageListenerAdapter messageListenerAdapter() { return new MessageListenerAdapter(logRabbitRecHandler, jsonMessageConverter()); } /** * 这个bean用于监听服务端发过来的消息，监听的Queue为logQueue()， * 因为该Queue绑定了logExchange和logRouteKey， 所以它可以接收到我们发送的日志消息 * @return */ @Bean public SimpleMessageListenerContainer messageListenerContainer() { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer( rabbit_connectionFactory()); container.setConcurrentConsumers(5); container.setQueues(logQueue()); container.setMessageListener(messageListenerAdapter()); container.setAcknowledgeMode(AcknowledgeMode.AUTO); return container; }}
6.封装发送消息的接口，如下所示。这是一个泛型的接口，目的是为了传入不同的消息类型。
package com.xdx.spring_rabbit;/** * 定义一个泛型接口，用于发送消息，T为要发送的消息类型 * @author xdx * * @param <T> */public interface RabbitSend<T> { void send(T t);}
7.实现这个发送消息的接口。在这个实现类中，我们注入了之前生成的RabbitTemplate对象。用于发送消息。
package com.xdx.spring_rabbit;import javax.annotation.Resource;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.stereotype.Component;import com.xdx.entity.TLog;/** * 用于发送日志消息的通用实现类 * * @author xdx * */@Component("logRabbitSend")public class LogRabbitSend implements RabbitSend<TLog> { @Resource(name = "rabbitTemplate") private RabbitTemplate rabbitTemplate; @Override public void send(TLog log) { rabbitTemplate.convertAndSend(log); System.err.println("发送消息：" + log); }}
8.封装监听器委托对象的接口，该接口用于处理监听器监听到的消息。同意是一个泛型的类，如下所示。
package com.xdx.spring_rabbit;/** * 用于处理监听到的消息的消息处理器接口，T为接收到的消息的类型 * * @author xdx * * @param <T> */public interface RabbitRecHandler<T> { void handleMessage(T t);}
9.实现上述委托对象的接口，如下所示。在该接口中，我们注入了日志处理类的对象。用于储存日志信息到数据库。
package com.xdx.spring_rabbit;import javax.annotation.Resource;import org.springframework.stereotype.Component;import com.xdx.entity.TLog;import com.xdx.service.LogService;@Component("logRabbitRecHandler")public class LogRabbitRecHandler implements RabbitRecHandler<TLog> { @Resource(name="logService") private LogService logService; @Override public void handleMessage(TLog log) { System.err.println("开始存储日志"+log.getOperator()+","+log.getEvent()); logService.saveLog(log); }}
10.最后，我们在具体的业务类中调用消息发送的接口，就可以实现日志消息的发送了。如下所示。
@Controllerpublic class AdminController { @Resource(name = "logRabbitSend") private LogRabbitSend logRabbitSend; @RequestMapping("admin") public ModelAndView admin(HttpSession session,String adminName, String password) throws Exception { List<Map<String,Object>>adminMap=adminService.getAllAdminMap(); ModelAndView mv = new ModelAndView(); //登录操作的主要逻辑代码…… session.setAttribute("adminName", admin.getAdminName()); session.setAttribute("realName", admin.getRealName()); TLog log=new TLog(adminName, "登录系统"); logRabbitSend.send(log); return mv; }}
运行我们的系统，我们先看看RabbitMQ的后台。看到了我们定义的Exchange和Queue等元素。
运行AdmintController类中的admin方法，登录系统，我们发现确实已经发送了消息，并且消息被监听到，然后存储到了数据库。
控制台打印出来的消息为：
数据库存入的记录为：
注：本文内容来自互联网，旨在为开发者提供分享、交流的平台。如有涉及文章版权等事宜，请你联系站长进行处理。