实现收到IBMMQ的数据之后存储到数据库(微服务系列第五天)

		<dependency>
			<groupId>org.mybatis.spring.boot</groupId>
			<artifactId>mybatis-spring-boot-starter</artifactId>
			<version>2.1.4</version>
		</dependency>
		<dependency>
			<groupId>mysql</groupId>
			<artifactId>mysql-connector-java</artifactId>
			<scope>runtime</scope>
		</dependency>
		<dependency>
			<groupId>com.alibaba</groupId>
			<artifactId>druid</artifactId>
			<version>1.2.4</version>
		</dependency>

# 只有下面三个是必填项（使用内嵌数据库的话这三个也可以不用填，会使用默认配置），其他配置不是必须的
spring.datasource.url=jdbc:mysql://172.25.0.2:3306/db1
spring.datasource.username=root
spring.datasource.password=123456
# driver-class-name 非必填可根据url推断
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver

# Druid 数据源配置，继承spring.datasource.* 配置，相同则覆盖
spring.datasource.initial-size=2
spring.datasource.max-active=30
spring.datasource.min-idle=2
spring.datasource.max-wait=1234
spring.datasource.pool-prepared-statements=true
spring.datasource.max-pool-prepared-statement-per-connection-size=5
# spring.datasource.max-open-prepared-statements= #等价于上面的max-pool-prepared-statement-per-connection-size
spring.datasource.validation-query=select 1
spring.datasource.validation-query-timeout=1
spring.datasource.test-on-borrow=true
spring.datasource.test-on-return=true
spring.datasource.test-while-idle=true
spring.datasource.time-between-eviction-runs-millis=10000
spring.datasource.min-evictable-idle-time-millis=30001
spring.datasource.async-close-connection-enable=true


# spring.datasource.aop-patterns=com.alibaba.spring.boot.demo.service.*

# 自定义StatFilter 配置 其他 Filter 不再演示
spring.datasource.filter.stat.db-type=h2
spring.datasource.filter.stat.log-slow-sql=true
spring.datasource.filter.stat.slow-sql-millis=2000

# JPA
spring.jpa.show-sql= true
spring.jpa.hibernate.ddl-auto=create-drop

# 配置下面参数用于启动监控页面，考虑安全问题，默认是关闭的，按需开启
spring.datasource.stat-view-servlet.enabled=true
spring.datasource.filter.stat.enabled=true
spring.datasource.web-stat-filter.enabled=true

# 更多配置属性见 DruidDataSource 内成员变量（只要有set方法便支持），或者根据IDE提示，或者查看官方文档

@Configuration
public class DruidConfig {

    /**
     * 数据源
     * 
     * @return
     */
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource")
    public DataSource druid() {
        return new DruidDataSource();
    }
}

    @Test
    public void testDataSourceExists() throws Exception {
        assertThat(dataSource).isNotNull();
    }

    @Test
    public void testDataSourcePropertiesOverridden() throws Exception {
        assertThat(dataSource.getUrl()).isEqualTo("jdbc:mysql://172.25.0.2:3306/db1");
        assertThat(dataSource.getInitialSize()).isEqualTo(2);
        assertThat(dataSource.getMaxActive()).isEqualTo(30);
        assertThat(dataSource.getMinIdle()).isEqualTo(2);
        assertThat(dataSource.getMaxWait()).isEqualTo(1234);
        assertThat(dataSource.isPoolPreparedStatements()).isTrue();
        //assertThat(dataSource.getMaxOpenPreparedStatements()).isEqualTo(5); //Duplicated with following
        assertThat(dataSource.getMaxPoolPreparedStatementPerConnectionSize()).isEqualTo(5);
        assertThat(dataSource.getValidationQuery()).isEqualTo("select 1");
        assertThat(dataSource.getValidationQueryTimeout()).isEqualTo(1);
        assertThat(dataSource.isTestWhileIdle()).isTrue();
        assertThat(dataSource.isTestOnBorrow()).isTrue();
        assertThat(dataSource.isTestOnReturn()).isTrue();
        assertThat(dataSource.getTimeBetweenEvictionRunsMillis()).isEqualTo(10000);
        assertThat(dataSource.getMinEvictableIdleTimeMillis()).isEqualTo(30001);
        assertThat(dataSource.isAsyncCloseConnectionEnable()).isEqualTo(true);
    }

public class MessageModel {
    /**
     * 消息ID
     */
    private long msgid;

    /**
     * 消息内容
     */
    private String content;

    /**
     * 消息保存到数据库的时间
     */
    private long created;

    public void setContent(String content) {
        this.content = content;
    }

    public void setCreated(long created) {
        this.created = created;
    }

    public void setMsgid(long msgid) {
        this.msgid = msgid;
    }

    public String getContent() {
        return content;
    }

    public long getCreated() {
        return created;
    }

    public long getMsgid() {
        return msgid;
    }

}

@Data
public class MessageData {
    /**
     * 消息ID
     */
    private long msgid;

    /**
     * 消息内容
     */
    private String content;

    /**
     * 消息保存到数据库的时间
     */
    private long created;

}

@Repository
public interface MessageMapper {
    /**
     * 从nsq的topic获取到数据保存到db中存储
     * @param msg
     */
    @Insert("insert into message (content,created) values (#{content},#{created})")
    void save(Message msg);

    /**
     * 从nsq的topic获取到数据保存到db中存储       测试@data写法数据的提交
     * @param msg
     */
    @Insert("insert into message (content,created) values (#{content},#{created})")
    void saveData(MessageData msg);
}

public interface MessageService {
    /**
     * 从nsq的topic获取到数据保存到db中存储
     */
    void save(Message msg);
    void saveData(MessageData msg);
}

@Component
public class MessageServiceImpl implements MessageService {

    @Autowired
    MessageMapper messageMapper;

    @Override
    public void save(Message msg) {
        // TODO Auto-generated method stub
        messageMapper.save(msg);
    }

    @Override
    public void saveData(MessageData msg) {
        // TODO Auto-generated method stub
        messageMapper.saveData(msg);
    }
}

@SpringBootApplication
@EnableTarsServer
@MapperScan("com.example.tarsmqserver.dao")
public class TarsMqServerApplication {

	public static void main(String[] args) {
		// 关闭 spring boot 自带的web服务 目前场景只用到了rpc服务
		SpringApplication app = new SpringApplication(TarsMqServerApplication.class);
		app.setWebApplicationType(WebApplicationType.NONE);
		app.run(args);
	}

}

    @Autowired
    MessageServiceImpl imp;

    @Test
    public void saveMessage() {
        Message message = new Message();
        message.setContent("abcdefg");
        message.setCreated(System.currentTimeMillis());
        imp.save(message);
    }

    @Autowired
    private NsqConfig nsqConfig;

    /**
     * 使用JmsListener配置消费者监听的队列
     * 
     * @param receivedMsg 接收到的消息
     */
    @JmsListener(destination = "#{@conf.recvQueueName}")
    public void receiveQueue(String receivedMsg) {
        JMS_LISTENER.info("RECEIVE: {}", receivedMsg);

        MessageModel messagObject = new MessageModel();
        messagObject.setContent(receivedMsg);
        messagObject.setCreated(System.currentTimeMillis());

        try {
            imp.save(messagObject);
        } catch (Exception e) {
            e.printStackTrace();
            JMS_LISTENER.error(e.toString());
            //TODO: handle exception
        }
    }

$ make build-upload