点击上方蓝字关注我们
上文:spring整合各种中间件(RocketMQ、kafka、RabbitMQ、TubeMQ、NSQ)-腾讯开源【TubeMQ】
本文源码下载:https://gitee.com/hong99/spring/issues/I1N1DF
NSQ是什么?
简介
NSQ 是一个实时分布式消息平台,旨在大规模运行,每天处理数十亿条消息。
它提倡没有单点故障的分布式和分散式拓扑结构,实现容错和高可用性,同时保证可靠的消息传递。请看特点和保证。
在操作上,NSQ很容易配置和部署(所有参数都在命令行上指定,编译的二进制文件没有运行时的依赖性)。为了获得最大的灵活性,它与数据格式无关(消息可以是JSON、MsgPack、协议缓冲区或其他任何形式)。官方的Go和Python库是开箱即用的(还有许多其他的客户端库),如果你有兴趣建立自己的库,有一个协议规范。
架构:
监控界面:
无单点故障:
特点:
详细请考这里:https://nsq.io/overview/design.html
源码实现
部署,请参考另一个文章:NSQ基于windows和docker的部署
java整合nsq
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>spring_mq</artifactId>
<groupId>com.hong</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.hong.nsq</groupId>
<artifactId>java_nsqmq</artifactId>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>8</source>
<target>8</target>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>com.sproutsocial</groupId>
<artifactId>nsq-j</artifactId>
<version>1.2</version>
</dependency>
</dependencies>
</project>
package com.hong.nsq;
/**
*
* 功能描述: 消费者
*
* @param:
* @return:
* @auther: csh
* @date: 2021/6/3 15:19
*/
import com.sproutsocial.nsq.Subscriber;
public class Consumer {
public static void handleData(byte[] data) {
System.out.println("接收到的消息:" + new String(data));
}
public static void main(String[] args) {
Subscriber subscriber = new Subscriber("localhost:4161");
subscriber.subscribe("nsq_topic", "hong_channel", Consumer::handleData);
}
}
package com.hong.nsq;
import com.sproutsocial.nsq.Publisher;
/**
* @author: csh
* @Date: 2021/6/3 15:17
* @Description:生产
*/
public class Publish {
public static void main(String[] args) {
Publisher publisher = new Publisher("localhost:4150");
byte[] data = "Hello nsq I'm hong.".getBytes();
publisher.publishBuffered("nsq_topic", data);
}
}
先启动消费,再启动生产!
结果:
spring 整合 nsq
消费者:spring_nsqmq_consumer
│ pom.xml
│ spring_nsqmq_consumer.iml
│
├─src
│ ├─main
│ │ ├─java
│ │ │ └─com
│ │ │ └─hong
│ │ │ └─spring
│ │ │ ├─config
│ │ │ │ NsqMqConsumer.java
│ │ │ │
│ │ │ ├─dao
│ │ │ │ UserMapper.java
│ │ │ │
│ │ │ ├─listener
│ │ │ │ ConsumerCommon.java
│ │ │ │ UserListener.java
│ │ │ │
│ │ │ ├─mapper
│ │ │ │ UserMapper.xml
│ │ │ │
│ │ │ └─provider
│ │ │ UserServiceImpl.java
│ │ │
│ │ └─resources
│ │ application.properties
│ │ applicationContext.xml
│ │ jdbc.properties
│ │ log4j2.xml
│ │ logging.properties
│ │ mybatis.xml
│ │ nsqmq.properties
│ │ nsqmq.xml
│ │
│ └─test
│ └─java
└─web
└─WEB-INF
web.xml
结构
源码实现
com.hong.spring.config.NsqMqConsumer
package com.hong.spring.config;
import com.hong.spring.listener.ConsumerCommon;
import com.sproutsocial.nsq.MessageDataHandler;
import com.sproutsocial.nsq.MessageHandler;
import com.sproutsocial.nsq.Subscriber;
import lombok.extern.log4j.Log4j2;
/**
* @author: csh
* @Date: 2021/4/27 18:06
* @Description:zero配置
*/
@Log4j2
public class NsqMqConsumer extends Subscriber{
//topic
private String topic;
//channel
private String channel;
//消息
private ConsumerCommon consumerCommon;
public NsqMqConsumer(String... lookupHosts) {
super(lookupHosts);
}
public void init(){
subscribe(topic,channel,consumerCommon::handleData);
}
@Override
public synchronized void subscribe(String topic, String channel, MessageDataHandler handler) {
super.subscribe(topic, channel, handler);
}
public String getTopic() {
return topic;
}
public void setTopic(String topic) {
this.topic = topic;
}
public String getChannel() {
return channel;
}
public void setChannel(String channel) {
this.channel = channel;
}
public ConsumerCommon getConsumerCommon() {
return consumerCommon;
}
public void setConsumerCommon(ConsumerCommon consumerCommon) {
this.consumerCommon = consumerCommon;
}
}
com.hong.spring.dao.UserMapper
package com.hong.spring.dao;
import com.hong.spring.entity.User;
import com.hong.spring.entity.ao.UserAO;
import org.apache.ibatis.annotations.Param;
import java.util.List;
/**
* @Auther: csh
* @Date: 2020/8/18 15:04
* @Description:用户dao层
*/
public interface UserMapper {
/**
*
* 功能描述:查询总条数
*
* @param:
* @return:
* @auther: csh
* @date: 2020/8/18 15:31
*/
List<User> findAllUserList();
/**
*
* 功能描述:获取总数
*
* @param:
* @return:
* @auther: csh
* @date: 2020/8/18 15:30
*/
int findAllTotal();
/**
*
* 功能描述:更新
*
* @param:
* @return:
* @auther: csh
* @date: 2020/8/18 15:30
*/
int update(User user);
/**
*
* 功能描述:添加
*
* @param:
* @return:
* @auther: csh
* @date: 2020/8/19 18:39
*/
int save(User user);
/**
*
* 功能描述:批量添加
*
* @param:
* @return:
* @auther: csh
* @date: 2020/8/21 15:46
*/
int insertBatch(@Param("list") List <User> list);
/**
*
* 功能描述:通过id查询
*
* @param:
* @return:
* @auther: csh
* @date: 2020/8/19 18:39
*/
User findById(int id);
/**
*
* 功能描述:通过分页查询
*
* @param:
* @return:
* @auther: csh
* @date: 2020/8/21 16:05
*/
List<User> findByPage(UserAO ao);
}
com.hong.spring.listener.ConsumerCommon
package com.hong.spring.listener;
/**
* @author: csh
* @Date: 2021/6/8 10:28
* @Description:公共接口
*/
public interface ConsumerCommon {
void handleData(byte[] data);
}
com.hong.spring.listener.UserListener
package com.hong.spring.listener;
import com.alibaba.fastjson.JSONObject;
import com.hong.spring.api.IUserService;
import com.hong.spring.entity.User;
import com.hong.spring.utils.DataResponse;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* @author: csh
* @Date: 2021/3/16 11:14
* @Description:用户监听
*/
@Log4j2
@Component
public class UserListener implements ConsumerCommon {
@Autowired
private IUserService userService;
@Override
public void handleData(byte[] data) {
if(null!=data && data.length>0){
String str = new String(data);
log.info("获取的用户信息{}", str);
User user = JSONObject.parseObject(str, User.class);
DataResponse <Boolean> save = userService.save(user);
if(save==null || save.getData()==null || !save.getData()){
log.info("添加失败,原因{}",JSONObject.toJSONString(save));
}
}else{
log.info("获取为空!");
}
}
}
com/hong/spring/mapper/UserMapper.xml
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.hong.spring.dao.UserMapper">
<resultMap type="com.hong.spring.entity.User" id="user">
<id column="id" property="id" />
<result column="user_name" property="username" />
<result column="age" property="age" />
</resultMap>
<select id="findById" resultType="com.hong.spring.entity.User">
SELECT * FROM user WHERE id = #{id,jdbcType=INTEGER}
</select>
<select id="findByPage" resultMap="user" parameterType="com.hong.spring.entity.ao.UserAO">
select * from user where 1=1 limit #{page},#{pageSize}
</select>
<select id="findAllUserList" resultMap="user">
SELECT * FROM user
</select>
<select id="findAllTotal" resultType="int">
SELECT count(*) FROM user
</select>
<insert id="save" >
INSERT INTO user ( user_name, age)
VALUES (#{username,jdbcType=VARCHAR},
#{age,jdbcType=INTEGER})
</insert>
<insert id="insertBatch">
insert into user
( user_name, age)
values
<foreach collection="list" item="user" index="index"
separator=",">
(#{user.username,jdbcType=VARCHAR},#{user.age,jdbcType=INTEGER})
</foreach>
</insert>
<update id="update" >
update user
<set>
<if test="username !=null">
user_name=#{username,jdbcType=VARCHAR},
</if>
<if test="age !=null">
age =#{age,jdbcType=INTEGER}
</if>
</set>
where id = #{id,jdbcType=INTEGER}
</update>
</mapper>
com.hong.spring.provider.UserServiceImpl
package com.hong.spring.provider;
import com.hong.spring.api.IUserService;
import com.hong.spring.dao.UserMapper;
import com.hong.spring.entity.User;
import com.hong.spring.entity.ao.UserAO;
import com.hong.spring.utils.DataResponse;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.List;
/**
* @Auther: csh
* @Date: 2020/8/18 15:16
* @Description:用户实现
*/
@Service("userService")
public class UserServiceImpl implements IUserService {
@Autowired
private UserMapper userDao;
@Override
public DataResponse<List<User>> findByAll() {
List <User> allUserList = userDao.findAllUserList();
int allTotal = userDao.findAllTotal();
return DataResponse.BuildSuccessResponse(allUserList,allTotal);
}
@Override
@Transactional
public DataResponse <Boolean> save(User user) {
if(null==user){
return DataResponse.BuildFailResponse("必传参数不能为空!");
}
int save = userDao.save(user);
return DataResponse.BuildSuccessResponse(save>0?true:false);
}
@Override
public DataResponse <Boolean> insertBatch(List <User> list) {
if(null==list){
return DataResponse.BuildFailResponse("参数不能为空!");
}
int batchSave = userDao.insertBatch(list);
return DataResponse.BuildSuccessResponse(batchSave>0?true:false);
}
@Override
@Transactional
public DataResponse <Boolean> update(User user) {
if(null==user || user.getId()==null){
return DataResponse.BuildFailResponse("必传参数不能为空!");
}
int update = userDao.update(user);
return DataResponse.BuildSuccessResponse(update>0?true:false);
}
@Override
public DataResponse <User> findById(int i) {
User byId = userDao.findById(i);
return DataResponse.BuildSuccessResponse(byId);
}
@Override
public DataResponse <List <User>> findByPage(UserAO ao) {
if(ao==null){
ao.setPage(0);
ao.setPageSize(10);
}else{
ao.setPage(ao.getPageSize() * ao.getPage());
}
int allTotal = userDao.findAllTotal();
List <User> byPage = userDao.findByPage(ao);
return DataResponse.BuildSuccessResponse(byPage,allTotal);
}
}
application.properties
logging.level.root=WARN
logging.level.org.springframework.web=DEBUG
logging.level.org.hibernate=ERROR
applicationContext.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:tx="http://www.springframework.org/schema/tx"
xmlns:mvc="http://www.springframework.org/schema/mvc"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd">
<!-- 配置组件扫描 -->
<context:component-scan base-package="com.hong.spring"></context:component-scan>
<!--加载配置文件-->
<context:property-placeholder location="classpath:jdbc.properties,classpath:nsqmq.properties"/>
<!-- 开启注解 -->
<context:annotation-config />
<!--开启注解事务-->
<tx:annotation-driven transaction-manager="transactionManager" />
<!--放行静态资源-->
<mvc:default-servlet-handler />
<bean class="org.springframework.web.servlet.view.InternalResourceViewResolver"
id="internalResourceViewResolver">
<!-- 前缀 -->
<property name="prefix" value="/WEB-INF/pages/" />
<!-- 后缀 -->
<property name="suffix" value=".html" />
<property name="contentType" value="text/html"/>
</bean>
<!--开启mvc注解事务-->
<!-- 定义注解驱动 -->
<mvc:annotation-driven>
<mvc:message-converters>
<!-- 设置支持中文 -->
<bean class="org.springframework.http.converter.StringHttpMessageConverter">
<property name="supportedMediaTypes">
<list>
<value>text/plain;charset=UTF-8</value>
<value>text/html;charset=UTF-8</value>
</list>
</property>
</bean>
<bean class="com.alibaba.fastjson.support.spring.FastJsonHttpMessageConverter"/>
</mvc:message-converters>
</mvc:annotation-driven>
<bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource">
<!-- 基础配置 -->
<property name="url" value="${jdbc.url}"></property>
<property name="driverClassName" value="${jdbc.driver}"></property>
<property name="username" value="${jdbc.user}"></property>
<property name="password" value="${jdbc.password}"></property>
<!-- 关键配置 -->
<!-- 初始化时建立物理连接的个数。初始化发生在显示调用init方法,或者第一次getConnection时 -->
<property name="initialSize" value="3" />
<!-- 最小连接池数量 -->
<property name="minIdle" value="2" />
<!-- 最大连接池数量 -->
<property name="maxActive" value="15" />
<!-- 配置获取连接等待超时的时间 -->
<property name="maxWait" value="10000" />
<!-- 性能配置 -->
<!-- 打开PSCache,并且指定每个连接上PSCache的大小 -->
<property name="poolPreparedStatements" value="true" />
<property name="maxPoolPreparedStatementPerConnectionSize" value="20" />
<!-- 其他配置 -->
<!-- 配置间隔多久才进行一次检测,检测需要关闭的空闲连接,单位是毫秒 -->
<property name="timeBetweenEvictionRunsMillis" value="60000" />
<!-- 配置一个连接在池中最小生存的时间,单位是毫秒 -->
<property name="minEvictableIdleTimeMillis" value="300000" />
<!-- 建议配置为true,不影响性能,并且保证安全性。申请连接的时候检测,如果空闲时间大于timeBetweenEvictionRunsMillis,
执行validationQuery检测连接是否有效。-->
<property name="testWhileIdle" value="true" />
<!-- 这里建议配置为TRUE,防止取到的连接不可用 ,申请连接时执行validationQuery检测连接是否有效,做了这个配置会降低性能。-->
<property name="testOnBorrow" value="true" />
<!-- 归还连接时执行validationQuery检测连接是否有效,做了这个配置会降低性能 -->
<property name="testOnReturn" value="false" />
</bean>
<!--事务管理器-->
<!-- sqlSessionFactory -->
<bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
<!-- 加载 MyBatis 的配置文件 -->
<property name="configLocation" value="classpath:mybatis.xml"/>
<!-- 数据源 -->
<property name="dataSource" ref="dataSource"/>
<!-- 所有配置的mapper文件 -->
<property name="mapperLocations" value="classpath*:com/hong/spring/mapper/*.xml" />
</bean>
<!-- Mapper 扫描器 -->
<bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
<!-- 扫描 包下的组件 -->
<property name="basePackage" value="com.hong.spring.dao" />
<!-- 关联mapper扫描器 与 sqlsession管理器 -->
<property name="sqlSessionFactoryBeanName" value="sqlSessionFactory" />
</bean>
<!--事务配置-->
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource" />
</bean>
</beans>
jdbc.properties
config.properties:
#数据库驱动
jdbc.driver=com.mysql.jdbc.Driver
#数据库连接url
jdbc.url=jdbc:mysql://localhost:3306/spring?useUnicode=true&characterEncoding=utf-8
#数据库用户名
jdbc.user=root
#数据库密码
jdbc.password=123456
log4j2.xml
<?xml version="1.0" encoding="UTF-8"?>
<configuration status="INFO">
<appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
</Console>
<RollingFile name="RollingFile" fileName="logs/app.log"
filePattern="logs/$${date:yyyy-MM}/app-%d{MM-dd-yyyy}-%i.log.gz">
<PatternLayout pattern="%d{yyyy.MM.dd 'at' HH:mm:ss z} %-5level %class{36} %L %M - %msg%xEx%n"/>
<SizeBasedTriggeringPolicy size="5 MB"/>
</RollingFile>
</appenders>
<loggers>
<root level="DEBUG">
<appender-ref ref="Console"/>
<appender-ref ref="RollingFile"/>
</root>
</loggers>
</configuration>
logging.properties
org.apache.catalina.core.ContainerBase.[Catalina].level=INFO
org.apache.catalina.core.ContainerBase.[Catalina].handlers=java.util.logging.ConsoleHandler
org.apache.jasper.servlet.TldScanner.level = FINE
handlers = org.apache.juli.FileHandler, java.util.logging.ConsoleHandler
############################################################
# Handler specific properties.
# Describes specific configuration info for Handlers.
############################################################
org.apache.juli.FileHandler.level = FINE
org.apache.juli.FileHandler.directory = ../logs
org.apache.juli.FileHandler.prefix = error-debug.
java.util.logging.ConsoleHandler.level = FINE
java.util.logging.ConsoleHandler.formatter = java.util.logging.SimpleFormatter
mybatis.xml
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration
PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>
<!-- settings -->
<settings>
<!-- 打开延迟加载的开关 -->
<setting name="lazyLoadingEnabled" value="true"/>
<!-- 将积极加载改为消极加载(即按需加载) -->
<setting name="aggressiveLazyLoading" value="false"/>
<!-- 打开全局缓存开关(二级缓存)默认值就是 true -->
<setting name="cacheEnabled" value="true"/>
<!-- 开启驼峰命名转换 Table(create_time) -> Entity(createtime) -->
<setting name="mapUnderscoreToCamelCase" value="true"/>
<!-- 使用列别名代替列名 默认:true seslect name as title from table -->
<setting name="useColumnLabel" value="true"/>
<!--使用jdbc的getGeneratedKeys获取数据库自增主键值-->
<setting name="useGeneratedKeys" value="true"/>
</settings>
<!-- 别名定义 -->
<typeAliases>
<package name="com.hong.spring.entity"/>
</typeAliases>
</configuration>
nsqmq.properties
nsqmq.url=localhost:4161
nsqmq.topic_user=nsq_spring_user
nsqmq.channel=hong_channel
nsqmq.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"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd">
<!--tubemq配置-->
<bean id="userSubscriber" class="com.hong.spring.config.NsqMqConsumer" init-method="init">
<constructor-arg name="lookupHosts" value="${nsqmq.url}"/>
<property name="topic" value="${nsqmq.topic_user}" />
<property name="channel" value="${nsqmq.channel}" />
<property name="consumerCommon" value="#{userListener}" />
</bean>
</beans>
WEB-INF/web.xml
<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"
version="3.1">
<servlet>
<servlet-name>spring_nsqmq_consumer</servlet-name>
<servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
<init-param>
<param-name>contextConfigLocation</param-name>
<param-value>classpath:applicationContext.xml,
classpath:nsqmq.xml
</param-value>
</init-param>
<load-on-startup>1</load-on-startup>
</servlet>
<filter>
<filter-name>encodingFilter</filter-name>
<filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
<init-param>
<param-name>encoding</param-name>
<param-value>UTF-8</param-value>
</init-param>
<init-param>
<param-name>forceEncoding</param-name>
<param-value>true</param-value>
</init-param>
</filter>
<filter-mapping>
<filter-name>encodingFilter</filter-name>
<url-pattern>/*</url-pattern>
</filter-mapping>
<servlet-mapping>
<servlet-name>spring_nsqmq_consumer</servlet-name>
<url-pattern>/</url-pattern>
</servlet-mapping>
</web-app>
spring_mq/spring_nsqmq_consumer/pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>spring_mq</artifactId>
<groupId>com.hong</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.hong.spring</groupId>
<artifactId>spring_nsqmq_consumer</artifactId>
<dependencies>
<dependency>
<artifactId>spring_mq_common_api</artifactId>
<version>1.0-SNAPSHOT</version>
<groupId>com.hong</groupId>
</dependency>
<dependency>
<groupId>com.sproutsocial</groupId>
<artifactId>nsq-j</artifactId>
<version>1.2</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.71</version>
</dependency>
</dependencies>
<build>
<resources>
<resource>
<directory>src/main/java</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
</resources>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
tomcat配置
生产者:spring_nsqmq_producer
│ pom.xml
│ spring_nsqmq_producer.iml
│
├─src
│ ├─main
│ │ ├─java
│ │ │ └─com
│ │ │ └─hong
│ │ │ └─spring
│ │ │ ├─config
│ │ │ │ CommonTopic.java
│ │ │ │ MsqMqProducer.java
│ │ │ │
│ │ │ └─controller
│ │ │ │ UserController.java
│ │ │ │
│ │ │ └─ao
│ │ │ UserSaveAO.java
│ │ │
│ │ └─resources
│ │ application.properties
│ │ applicationContext.xml
│ │ log4j2.xml
│ │ logging.properties
│ │ nsqmq.properties
│ │ nsqmq.xml
│ │
│ └─test
│ └─java
└─web
└─WEB-INF
web.xml
源码实现
spring_mq/spring_nsqmq_producer/pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>spring_mq</artifactId>
<groupId>com.hong</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.hong.spring</groupId>
<artifactId>spring_nsqmq_producer</artifactId>
<dependencies>
<dependency>
<artifactId>spring_mq_common_api</artifactId>
<version>1.0-SNAPSHOT</version>
<groupId>com.hong</groupId>
</dependency>
<dependency>
<groupId>com.sproutsocial</groupId>
<artifactId>nsq-j</artifactId>
<version>1.2</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.71</version>
</dependency>
</dependencies>
<build>
<resources>
<resource>
<directory>src/main/java</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
</resources>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
WEB-INF/web.xml
<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"
version="3.1">
<servlet>
<servlet-name>spring_nsqmq_producer</servlet-name>
<servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
<init-param>
<param-name>contextConfigLocation</param-name>
<param-value>classpath:applicationContext.xml,
classpath:nsqmq.xml
</param-value>
</init-param>
<load-on-startup>1</load-on-startup>
</servlet>
<filter>
<filter-name>encodingFilter</filter-name>
<filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
<init-param>
<param-name>encoding</param-name>
<param-value>UTF-8</param-value>
</init-param>
<init-param>
<param-name>forceEncoding</param-name>
<param-value>true</param-value>
</init-param>
</filter>
<filter-mapping>
<filter-name>encodingFilter</filter-name>
<url-pattern>/*</url-pattern>
</filter-mapping>
<servlet-mapping>
<servlet-name>spring_nsqmq_producer</servlet-name>
<url-pattern>/</url-pattern>
</servlet-mapping>
</web-app>
nsqmq.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"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd">
<!--msq配置-->
<bean id="push" class="com.hong.spring.config.MsqMqProducer" init-method="init" destroy-method="close">
<constructor-arg name="url" value="${msq.url}" />
</bean>
</beans>
nsqmq.properties
msq.url=localhost:4150
logging.properties
org.apache.catalina.core.ContainerBase.[Catalina].level=INFO
org.apache.catalina.core.ContainerBase.[Catalina].handlers=java.util.logging.ConsoleHandler
handlers = org.apache.juli.FileHandler, java.util.logging.ConsoleHandler
############################################################
# Handler specific properties.
# Describes specific configuration info for Handlers.
############################################################
org.apache.juli.FileHandler.level = FINE
org.apache.juli.FileHandler.directory = ../logs
org.apache.juli.FileHandler.prefix = error-debug.
java.util.logging.ConsoleHandler.level = FINE
java.util.logging.ConsoleHandler.formatter = java.util.logging.SimpleFormatter
log4j2.xml
<?xml version="1.0" encoding="UTF-8"?>
<configuration status="INFO">
<appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
</Console>
<RollingFile name="RollingFile" fileName="logs/app.log"
filePattern="logs/$${date:yyyy-MM}/app-%d{MM-dd-yyyy}-%i.log.gz">
<PatternLayout pattern="%d{yyyy.MM.dd 'at' HH:mm:ss z} %-5level %class{36} %L %M - %msg%xEx%n"/>
<SizeBasedTriggeringPolicy size="5 MB"/>
</RollingFile>
</appenders>
<loggers>
<root level="DEBUG">
<appender-ref ref="Console"/>
<appender-ref ref="RollingFile"/>
</root>
</loggers>
</configuration>
applicationContext.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:tx="http://www.springframework.org/schema/tx"
xmlns:mvc="http://www.springframework.org/schema/mvc"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd">
<!-- 配置组件扫描 -->
<context:component-scan base-package="com.hong.spring"></context:component-scan>
<!--加载配置文件-->
<context:property-placeholder location="classpath:nsqmq.properties"/>
<!-- 开启注解 -->
<context:annotation-config />
<mvc:default-servlet-handler />
<bean class="org.springframework.web.servlet.view.InternalResourceViewResolver"
id="internalResourceViewResolver">
<!-- 前缀 -->
<property name="prefix" value="/WEB-INF/pages/" />
<!-- 后缀 -->
<property name="suffix" value=".html" />
<property name="contentType" value="text/html"/>
</bean>
<!--开启mvc注解事务-->
<!-- 定义注解驱动 -->
<mvc:annotation-driven>
<mvc:message-converters>
<!-- 设置支持中文 -->
<bean class="org.springframework.http.converter.StringHttpMessageConverter">
<property name="supportedMediaTypes">
<list>
<value>text/plain;charset=UTF-8</value>
<value>text/html;charset=UTF-8</value>
</list>
</property>
</bean>
<bean class="com.alibaba.fastjson.support.spring.FastJsonHttpMessageConverter"/>
</mvc:message-converters>
</mvc:annotation-driven>
</beans>
application.properties
logging.level.root=WARN
logging.level.org.springframework.web=DEBUG
logging.level.org.hibernate=ERROR
com.hong.spring.controller.ao.UserSaveAO
package com.hong.spring.controller.ao;
import lombok.Data;
import java.io.Serializable;
/**
* @author: csh
* @Date: 2021/3/16 11:21
* @Description:用户入参
*/
@Data
public class UserSaveAO implements Serializable {
private Integer id;
private String username;
private Integer age;
}
com.hong.spring.controller.UserController
package com.hong.spring.controller;
import com.alibaba.fastjson.JSONObject;
import com.hong.spring.config.CommonTopic;
import com.hong.spring.config.MsqMqProducer;
import com.hong.spring.controller.ao.UserSaveAO;
import com.hong.spring.utils.DataResponse;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @Auther: csh
* @Date: 2020/8/18 16:11
* @Description:
*/
@RestController
@RequestMapping("/user/")
@Log4j2
public class UserController {
@Autowired
private MsqMqProducer push;
@RequestMapping("save")
public DataResponse<Boolean> save(UserSaveAO ao){
log.info("添加用户入参{}",JSONObject.toJSONString(ao));
if(null==ao){
return DataResponse.BuildFailResponse("参数不能为空!");
}
try {
Boolean send = push.send(JSONObject.toJSONString(ao),CommonTopic.NSQ_USER_TOPIC);
if(null==send || !send){
return DataResponse.BuildFailResponse("添加用户失败!");
}
return DataResponse.BuildFailResponse("添加用户成功!");
}catch (Exception e){
log.error("添加出错{}",e);
return DataResponse.BuildFailResponse("添加出错请重试!");
}
}
}
com.hong.spring.config.MsqMqProducer
package com.hong.spring.config;
import com.sproutsocial.nsq.Publisher;
import lombok.extern.log4j.Log4j2;
/**
* @author: csh
* @Date: 2021/4/27 18:06
* @Description:tubemq配置
*/
@Log4j2
public class MsqMqProducer {
/**服务端 */
private static Publisher pushlisher;
/**服务地址 */
private String url;
public MsqMqProducer(String url) {
this.url = url;
}
public void init () {
pushlisher = new Publisher(url);
}
/**
*
* 功能描述: 发送消息
*
* @param:
* @return:
* @auther: csh
* @date: 2021/4/28 17:05
*/
public Boolean send(String msg,String topic) {
try {
pushlisher.publishBuffered(topic,msg.getBytes());
return true;
}catch (Exception e){
log.error("发送失败{}",e);
}
return false;
}
public String getUrl() {
return url;
}
public void setUrl(String url) {
this.url = url;
}
/**
*
* 功能描述: 关闭
*
* @param:
* @return:
* @auther: csh
* @date: 2021/5/18 14:50
*/
public void close(){
try {
if(null!= pushlisher){
pushlisher.stop();
}
}catch (Exception e){
log.error("关闭失败{}",e);
} catch (Throwable throwable) {
log.error("关闭失败{}",throwable);
throwable.printStackTrace();
}
}
}
com.hong.spring.config.CommonTopic
package com.hong.spring.config;
/**
* @author: csh
* @Date: 2021/5/18 10:56
* @Description:公共的topic
*/
public class CommonTopic {
//用户mq
public static final String NSQ_USER_TOPIC ="nsq_spring_user";
}
然后请求地址:http://localhost:8785/user/save?username=spring_nsqmq&age=1
username:spring_nsqmq
age:1
结果
[2021-06-08 11:58:37,357] Artifact spring_nsqmq_producer:war exploded: Artifact is deployed successfully
[2021-06-08 11:58:37,357] Artifact spring_nsqmq_producer:war exploded: Deploy took 9,218 milliseconds
12:01:55.120 [http-nio-8785-exec-8] INFO com.hong.spring.controller.UserController - 添加用户入参{"age":1,"username":"spring_nsqmq"}
12:01:55.514 [nsq-batch-0] DEBUG com.sproutsocial.nsq.Connection - serverConfig:{"max_rdy_count":2500,"version":"1.2.0","max_msg_timeout":900000,"msg_timeout":60000,"tls_v1":false,"deflate":false,"deflate_level":6,"max_deflate_level":6,"snappy":false,"sample_rate":0,"auth_required":false,"output_buffer_size":16384,"output_buffer_timeout":250}
12:01:55.515 [nsq-batch-0] INFO com.sproutsocial.nsq.Connection - connected localhost:4150 msgTimeout:60000 heartbeatInterval:30000 maxRdyCount:2500
12:01:55.521 [nsq-batch-0] INFO com.sproutsocial.nsq.Publisher - publisher connected:localhost:4150
数据库记录
springboot整合nsq
通过springboot+dubbo+zk+nsq进行整合
服务 | dubbo端口 | 服务端口 |
---|---|---|
springboot_nsq_api | 20880 | 8586 |
springboot_nsq_consumer | 20882 | 8588 |
springboot_nsq_producer | 20881 | 8587 |
springboot_nsq_api
结构
│ pom.xml
│
└─src
└─main
├─java
│ └─com
│ └─hong
│ └─springboot
│ │ Application.java
│ │
│ └─controller
│ IndexController.java
│ UserController.java
│
└─resources
application.properties
springboot_all/springboot_nsq_api/pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>com.hong.springboot</groupId>
<artifactId>springboot_all</artifactId>
<version>0.0.1-SNAPSHOT</version>
<relativePath/>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.hong.springboot</groupId>
<artifactId>springboot_nsq_api</artifactId>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>com.hong.springboot</groupId>
<artifactId>springboot_mq_api</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.alibaba.boot</groupId>
<artifactId>dubbo-spring-boot-starter</artifactId>
<version>0.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.5.4-beta</version>
<exclusions>
<exclusion>
<artifactId>slf4j-api</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
<!--静态资源导出问题-->
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
<resources>
<resource>
<directory>src/main/java</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
</resources>
</build>
</project>
application.properties
#dubbo configuration
#服务名称
dubbo.application.name=springboot_dubbo_consumer
#注册中心协议
dubbo.registry.protocol=zookeeper
#注册地址
dubbo.registry.address=zookeeper://127.0.0.1:2181
#扫描注解包通过该设置将服务注册到zookeeper
dubbo.protocol.scan=com.hong.springboot.api
#注册端口
dubbo.protocol.port=20880
#协议名称
dubbo.protocol.name=dubbo
#扫包
dubbo.scan.basePackages=com.hong.springboot.controller
#避免端口冲突
server.port=8586
com.hong.springboot.Application
package com.hong.springboot;
import com.alibaba.dubbo.config.spring.context.annotation.EnableDubbo;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author: csh
* @Date: 2020/11/21 11:37
* @Description:springboot dubbo消费端
*/
@SpringBootApplication(scanBasePackages = "com.hong.springboot")
@EnableDubbo
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class);
}
}
com.hong.springboot.controller.UserController
package com.hong.springboot.controller;
import com.alibaba.dubbo.common.utils.StringUtils;
import com.alibaba.dubbo.config.annotation.Reference;
import com.hong.springboot.api.IUserService;
import com.hong.springboot.entity.User;
import com.hong.springboot.utils.DataResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
* @Auther: csh
* @Date: 2020/8/18 16:11
* @Description:用户
*/
@RestController
@Slf4j
@RequestMapping("/user")
public class UserController {
@Reference
private IUserService userService;
@GetMapping("/findByAll")
public DataResponse<List<User>> findByAll(){
try {
return userService.findByAll();
} catch (Exception e){
log.error("查询出错{}",e);
}
return DataResponse.BuildFailResponse("查询出错!");
}
@PostMapping("/save")
public DataResponse<Boolean> save(User ao){
if(null==ao || ao.getAge()==null || StringUtils.isBlank(ao.getUsername())){
return DataResponse.BuildFailResponse("参数不能为空!");
}
DataResponse <Boolean> save = userService.save(ao);
return save;
}
}
com.hong.springboot.controller.IndexController
package com.hong.springboot.controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @author: csh
* @Date: 2021/1/12 10:16
* @Description:首页
*/
@RestController
public class IndexController {
@RequestMapping("/")
public String index(){
return "成功!";
}
}
springboot_nsq_consumer
结构
│ pom.xml
│
└─src
└─main
├─java
│ └─com
│ └─hong
│ └─springboot
│ │ Application.java
│ │
│ ├─config
│ │ MessageCallback.java
│ │ NsqConsumer.java
│ │
│ └─listener
│ UserListener.java
│
└─resources
application.properties
com.hong.springboot.config.MessageCallback
package com.hong.springboot.config;
import java.io.Serializable;
public interface MessageCallback<T extends Serializable> {
void onMessage(T msg);
}
com.hong.springboot.Application
package com.hong.springboot;
import com.alibaba.dubbo.config.spring.context.annotation.EnableDubbo;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author: csh
* @Date: 2020/11/21 11:37
* @Description:
*/
@SpringBootApplication(scanBasePackages = "com.hong.springboot")
@EnableDubbo
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class);
}
}
com.hong.springboot.listener.UserListener
package com.hong.springboot.listener;
import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSONObject;
import com.hong.springboot.api.IUserService;
import com.hong.springboot.config.MessageCallback;
import com.hong.springboot.entity.User;
import com.hong.springboot.utils.DataResponse;
import lombok.extern.log4j.Log4j2;
import org.springframework.stereotype.Service;
import java.io.Serializable;
/**
* @author: csh
* @Date: 2021/3/16 11:14
* @Description:用户监听
*/
@Service
@Log4j2
public class UserListener implements MessageCallback<byte[]>, Serializable {
@Reference
private IUserService userService;
@Override
public void onMessage(byte[] data) {
if(null==data || data.length<1){
return;
}
String msg = new String(data);
try {
User user = JSONObject.parseObject(msg,User.class);
log.info("最终要添加的值{}",JSONObject.toJSONString(user));
DataResponse<Boolean> save = userService.reallySave(user);
if(save==null || !save.getData()){
log.info("添加失败,原因{}",JSONObject.toJSONString(save));
}
}catch (Exception e){
log.error("添加出错",e);
}
}
}
com.hong.springboot.config.NsqConsumer
package com.hong.springboot.config;
import com.sproutsocial.nsq.Subscriber;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PostConstruct;
@Configuration
public class NsqConsumer {
//topic
@Value("${springboot.nsqmq.topic_user}")
private String topic;
//channel
@Value("${springboot.nsqmq.channel}")
private String channel;
@Value("${springboot.nsqmq.host}")
private String lookupHosts;
//消息费
@Autowired
private MessageCallback messageCallback;
private Subscriber subscriber;
@PostConstruct
public void init(){
subscriber = new Subscriber(lookupHosts);
subscriber.subscribe(topic,channel,messageCallback::onMessage);
}
}
application.properties
rocketmq.name-server=localhost:9876
rocketmq.producer.group=hong_group
rocketmq.producer.sendMessageTimeout=300000
#避免端口冲突
server.port=8588
#dubbo configuration
#服务名称
dubbo.application.name=springboot_rocketmq_consumer
dubbo.registry.protocol=zookeeper
#注册地址
dubbo.registry.address=zookeeper://127.0.0.1:2181
#扫描注解包通过该设置将服务注册到zookeeper
dubbo.protocol.scan=com.hong.springboot.api
#注册端口
dubbo.protocol.port=20882
#协议名称
dubbo.protocol.name=dubbo
#扫包
dubbo.scan.basePackages=com.hong.springboot.listener
springboot.nsqmq.host=localhost:4161
springboot.nsqmq.topic_user=nsq_springboot_user
springboot.nsqmq.channel=hong_channel
springboot_all/springboot_nsq_consumer/pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>com.hong.springboot</groupId>
<artifactId>springboot_all</artifactId>
<version>0.0.1-SNAPSHOT</version>
<relativePath/>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.hong.springboot</groupId>
<artifactId>springboot_nsq_consumer</artifactId>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>com.sproutsocial</groupId>
<artifactId>nsq-j</artifactId>
<version>1.2</version>
</dependency>
<dependency>
<groupId>com.hong.springboot</groupId>
<artifactId>springboot_mq_api</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.boot</groupId>
<artifactId>dubbo-spring-boot-starter</artifactId>
<version>0.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.5.4-beta</version>
<exclusions>
<exclusion>
<artifactId>slf4j-api</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<!--<build>-->
<!--<plugins>-->
<!--<plugin>-->
<!--<groupId>org.springframework.boot</groupId>-->
<!--<artifactId>spring-boot-maven-plugin</artifactId>-->
<!--<configuration>-->
<!--<skip>true</skip>-->
<!--</configuration>-->
<!--</plugin>-->
<!--</plugins>-->
<!--</build>-->
<!--静态资源导出问题-->
<build>
<resources>
<resource>
<directory>src/main/java</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
</resources>
</build>
</project>
springboot_nsq_producer
结构
│ pom.xml
│
└─src
├─main
│ ├─java
│ │ └─com
│ │ └─hong
│ │ └─springboot
│ │ │ Application.java
│ │ │
│ │ ├─config
│ │ │ DruidConfig.java
│ │ │ NsqProducer.java
│ │ │ TopicList.java
│ │ │
│ │ ├─dao
│ │ │ UserMapper.java
│ │ │
│ │ └─provider
│ │ UserServiceImpl.java
│ │
│ └─resources
│ │ application.properties
│ │
│ └─META-INF
│ spring.factories
│
└─test
└─java
└─com
└─hong
└─springboot
MqTest.java
springboot_all/springboot_nsq_producer/pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>com.hong.springboot</groupId>
<artifactId>springboot_all</artifactId>
<version>0.0.1-SNAPSHOT</version>
<relativePath/>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.hong.springboot</groupId>
<artifactId>springboot_nsq_producer</artifactId>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-zeromq</artifactId>
<version>5.4.6</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.hong.springboot</groupId>
<artifactId>springboot_mq_api</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>2.1.3</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid-spring-boot-starter</artifactId>
<version>1.1.10</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.boot</groupId>
<artifactId>dubbo-spring-boot-starter</artifactId>
<version>0.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.5.4-beta</version>
<exclusions>
<exclusion>
<artifactId>slf4j-api</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.sproutsocial</groupId>
<artifactId>nsq-j</artifactId>
<version>1.2</version>
</dependency>
</dependencies>
<!--<build>-->
<!--<plugins>-->
<!--<plugin>-->
<!--<groupId>org.springframework.boot</groupId>-->
<!--<artifactId>spring-boot-maven-plugin</artifactId>-->
<!--<configuration>-->
<!--<skip>true</skip>-->
<!--</configuration>-->
<!--</plugin>-->
<!--</plugins>-->
<!--</build>-->
<!--静态资源导出问题-->
<build>
<resources>
<resource>
<directory>src/main/java</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
</resources>
</build>
</project>
application.properties
rocketmq.name-server=localhost:9876
rocketmq.producer.group=hong_group
rocketmq.producer.sendMessageTimeout=300000
#dubbo configuration
#服务名称
dubbo.application.name=springboot_rabbitmq_producer
dubbo.registry.protocol=zookeeper
#注册地址
dubbo.registry.address=zookeeper://127.0.0.1:2181
#扫描注解包通过该设置将服务注册到zookeeper
dubbo.protocol.scan=com.hong.springboot.api
#注册端口
dubbo.protocol.port=20881
#协议名称
dubbo.protocol.name=dubbo
#避免端口冲突
server.port=8587
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost:3306/spring?useUnicode=true&characterEncoding=utf-8&serverTimezone=UTC
spring.datasource.username=root
spring.datasource.password=123456
#mybatis配置
mybatis.typeAliasesPackage=com.hong.springboot.entity
#nsq配置
springboot.msq.url=localhost:4150
com.hong.springboot.Application
package com.hong.springboot;
import com.alibaba.dubbo.config.spring.context.annotation.EnableDubbo;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author: csh
* @Date: 2020/11/21 11:37
* @Description:
*/
@SpringBootApplication(scanBasePackages = "com.hong.springboot")
@MapperScan("com.hong.springboot.dao")
@EnableDubbo
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class);
}
}
com.hong.springboot.provider.UserServiceImpl
package com.hong.springboot.provider;
import com.alibaba.druid.util.StringUtils;
import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSONObject;
import com.hong.springboot.api.IUserService;
import com.hong.springboot.config.NsqProducer;
import com.hong.springboot.config.TopicList;
import com.hong.springboot.dao.UserMapper;
import com.hong.springboot.entity.User;
import com.hong.springboot.utils.DataResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import java.util.List;
/**
* @Auther: csh
* @Date: 2020/8/18 15:16
* @Description:用户实现
*/
@Service(interfaceClass = IUserService.class,timeout = 6000)
@Slf4j
public class UserServiceImpl implements IUserService {
@Autowired
private UserMapper userDao;
@Autowired
private NsqProducer nsqProducer;
@Override
public DataResponse<List<User>> findByAll() {
List <User> allUserList = userDao.findAllUser();
return DataResponse.BuildSuccessResponse(allUserList,allUserList.size());
}
//随机用
private static int i = 0;
@Override
public DataResponse <Boolean> save(User userAO) {
i++;
log.info("需要nsq添加的用户信息{}",JSONObject.toJSONString(userAO));
try {
//当==0时则用队列发送,当不等于的时候用topic发送。
nsqProducer.send(JSONObject.toJSONString(userAO),TopicList.USER_TOPIC);
}catch (Exception e){
e.printStackTrace();
return DataResponse.BuildSuccessResponse(false);
}
return DataResponse.BuildSuccessResponse(true);
}
@Transactional
@Override
public DataResponse <Boolean> reallySave(User user) {
log.info("要添加的用户信息{}",JSONObject.toJSONString(user));
if(null==user || user.getAge()==null || StringUtils.isEmpty(user.getUsername())){
return DataResponse.BuildFailResponse("参数不能为空!");
}
int insert = userDao.insert(user);
return insert>0?DataResponse.BuildSuccessResponse(true):DataResponse.BuildFailResponse("失败",false);
}
}
com.hong.springboot.dao.UserMapper
package com.hong.springboot.dao;
import com.hong.springboot.entity.User;
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Select;
import java.util.List;
/**
* @Auther: csh
* @Date: 2020/8/18 15:04
* @Description:用户dao层
*/
public interface UserMapper {
@Select("select id,user_name,age from user")
List<User> findAllUser();
@Insert("insert into user (user_name,age) values(#{username},#{age})")
int insert(User user);
}
com.hong.springboot.config.TopicList
package com.hong.springboot.config;
/**
* @author: csh
* @Date: 2021/6/9 13:55
* @Description:
*/
public class TopicList {
public static final String USER_TOPIC="nsq_springboot_user";
}
com.hong.springboot.config.NsqProducer
package com.hong.springboot.config;
import com.sproutsocial.nsq.Publisher;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
@Service
@Slf4j
public class NsqProducer {
@Value("${springboot.msq.url}")
private String url;
private Publisher publisher;
@PostConstruct
public void init(){
publisher = new Publisher(url);
}
@PreDestroy
public void close(){
try {
if(null!= publisher){
publisher.stop();
}
}catch (Exception e){
log.error("关闭失败{}",e);
} catch (Throwable throwable) {
log.error("关闭失败{}",throwable);
throwable.printStackTrace();
}
}
public Boolean send(String msg,String topic) {
try {
publisher.publishBuffered(topic,msg.getBytes());
return true;
}catch (Exception e){
log.error("发送失败{}",e);
}
return false;
}
}
com.hong.springboot.config.DruidConfig
package com.hong.springboot.config;
import com.alibaba.druid.pool.DruidDataSource;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.sql.DataSource;
/**
* @author: csh
* @Date: 2021/1/8 18:08
* @Description:数据源配置
*/
@Configuration
public class DruidConfig {
@Bean
@ConfigurationProperties(prefix = "spring.datasource")
public DataSource dataSource(){
return new DruidDataSource();
}
}
然后通过postman进行测试如下:
http://localhost:8586/user/save?username=springboot_nsq&age=10001
结果如下:
2021-06-09 14:32:45.026 INFO 12880 --- [:20881-thread-6] c.h.springboot.provider.UserServiceImpl : 需要nsq添加的用户信息{"age":10001,"username":"springboot_nsq"}
2021-06-09 14:32:45.343 INFO 12880 --- [ nsq-batch-0] com.sproutsocial.nsq.Connection : connected localhost:4150 msgTimeout:60000 heartbeatInterval:30000 maxRdyCount:2500
2021-06-09 14:32:45.344 INFO 12880 --- [ nsq-batch-0] com.sproutsocial.nsq.Publisher : publisher connected:localhost:4150
2021-06-09 14:34:41.307 INFO 12880 --- [20881-thread-10] c.h.springboot.provider.UserServiceImpl : 需要nsq添加的用户信息{"age":10001,"username":"springboot_nsq"}
2021-06-09 14:35:15.001 INFO 12880 --- [20881-thread-11] com.alibaba.druid.pool.DruidDataSource : {dataSource-1} inited
2021-06-09 14:35:15.515 INFO 12880 --- [20881-thread-11] c.h.springboot.provider.UserServiceImpl : 要添加的用户信息{"age":10001,"username":"springboot_nsq"}
2021-06-09 14:35:48.252 INFO 12880 --- [20881-thread-12] c.h.springboot.provider.UserServiceImpl : 要添加的用户信息{"age":10001,"username":"springboot_nsq"}
数据库结果
最后
nsq虽然说性能很优秀,但是有两点无法满足,第一个是消费是无序的,没办法保证有序,需要自已实现;第二个是消费会重复,最少会被投递一次,这点跟rocketmq有点类似,解决方案也相当简单,做幂等就可以了。至于文档来说相对来说nsq的文档还是比较全,国内[有赞]在用使场景,相对来说比起其他的中间件还是有中厂使用过的场景的,比较靠谱。
参考文章
https://nsq.io/overview/design.html
https://tech.youzan.com/building-nsq-client-cont/
https://www.infoq.cn/article/tVILYgR8iP31V3GAkxHZ
https://tech.youzan.com/building-nsq-client-cont/
至此,市面上大部分相关的中间件都已经整合完毕,可以参阅如下