message_bus / Message-trans · 提交

转到一个项目

作者 朱兆平
提交 4b97894d5ac064326e0dcb9e26ea958e9022030c 0 个父辈

init

内嵌 并排对比
正在显示 8 个修改的文件 包含 1123 行增加 和 0 行删除
  1 +/logs/
  2 +/target/
  3 +/out/
  4 +.idea/
  5 +messagebus-trans-message.iml
  6 +src/test/
  7 +target/
  1 +
  2 +#上传文件的路径，要带斜杠
  3 +web:
  4 + upload-path: upload/
  5 +server:
  6 + port: 11112
  7 + servlet:
  8 + context-path: ${SERVER_CONTEXTPATH:}
  9 +spring:
  10 + devtools:
  11 + add-properties: false
  12 + profiles:
  13 + active: dev
  14 + mvc:
  15 + #静态资源，设置上传文件的访问，
  16 + static-path-pattern: /**
  17 + resources:
  18 + static-locations: classpath:/META-INF/resources/,classpath:/static,classpath:/resources/,file:${web.upload-path}
  19 +
  20 + application:
  21 + name: message-bus-trans-message
  22 + redis:
  23 +# host: 127.0.0.1
  24 + host: 8.131.245.248
  25 + port: 6379
  26 + password: dsly@admin
  27 + jackson:
  28 + serialization:
  29 + FAIL_ON_EMPTY_BEANS: false
  30 + #springboot2.0之后会把Date类型字段自动给转成UTC字符串 如：1990-11-26T16:00:00.000+0000，如果想转成时间戳在application.properties配置文件增加以下配置
  31 + date-format: yyyy-MM-dd HH:mm:ss
  32 + #时区必须要设置
  33 + time-zone: GMT+8
  34 + #ALWAYS的意思是即时属性为null，仍然也会输出这个key
  35 + default-property-inclusion: always
  36 +
  37 + cloud:
  38 + #eureka主机名，会在控制页面中显示
  39 + #DEV环境关闭注册。
  40 + features:
  41 + enabled: true
  42 + discovery:
  43 + enabled: true
  44 + service-registry:
  45 + auto-registration:
  46 + enabled: true
  47 +
  48 + datasource:
  49 + type: com.alibaba.druid.pool.DruidDataSource
  50 + #oracle
  51 + #driver-class-name=oracle.jdbc.OracleDriver
  52 + #url=jdbc:oracle:thin:@10.50.3.68:1521:CGODW
  53 + #username=CGOETL
  54 + #password=1q2w3e4r
  55 + #spring datasource mysql，注意编码配置，缺少数据库编码配置容易引起中文入库乱码
  56 +# url: jdbc:mysql://118.31.66.166:3306/station?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8
  57 + url: jdbc:mysql://8.131.245.248:3306/zz_logistics?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8
  58 + username: root
  59 + password: ZZdsly123
  60 + driver-class-name: com.mysql.cj.jdbc.Driver
  61 + #配置初始化大小/最小/最大
  62 + druid:
  63 + initial-size: 10
  64 + min-idle: 10
  65 + max-active: 20
  66 + #获取连接等待超时时间
  67 + max-wait: 60000
  68 + #一个连接在池中最小生存的时间
  69 + min-evictable-idle-time-millis: 300000
  70 + #间隔多久进行一次检测，检测需要关闭的空闲连接
  71 + time-between-eviction-runs-millis: 60000
  72 + #mysql
  73 + validation-query: SELECT 1 FROM DUAL
  74 + #oracle
  75 + #druid.validation-query:SELECT 'x' FROM DUAL
  76 + test-while-idle: true
  77 + test-on-borrow: false
  78 + test-on-return: false
  79 + default-auto-commit: true
  80 +# tomcat:
  81 +# max-wait: 10
  82 +# max-idle: 10000
  83 +# min-idle: 5
  84 +# initial-size: 5
  85 +
  86 +eureka:
  87 + instance:
  88 + #eureka服务器页面中status的请求路径
  89 +# status-page-url: http://${eureka.instance.hostname}:${server.port}/index
  90 + prefer-ip-address: true
  91 + instance-id: ${spring.cloud.client.ip-address}:${server.port}
  92 + hostname: ${spring.cloud.client.ip-address}
  93 + lease-renewal-interval-in-seconds: 15
  94 + lease-expiration-duration-in-seconds: 45
  95 + health-check-url-path: /actuator/health
  96 + metadata-map:
  97 + user:
  98 + name: "admin" #These two are needed so that the server
  99 + password: "123456"
  100 + client:
  101 + #eureka注册中心服务器地址
  102 + service-url:
  103 +# defaultZone: http://127.0.0.1:12345/eureka/
  104 + defaultZone: http://192.168.1.53:12345/eureka/
  105 + registry-fetch-interval-seconds: 30
  106 +
  107 +
  108 +
  109 +management:
  110 + endpoints:
  111 + enabled-by-default: true
  112 + web:
  113 + exposure:
  114 + include: "*"
  115 + endpoint:
  116 + health:
  117 + show-details: always
  118 + shutdown:
  119 + enabled: true
  120 +
  121 +mybatis:
  122 + mapper-locations: classpath:mapping/*.xml
  123 + type-aliases-package: com.sunyo.usercenter.heartbeat.model
  124 +pagehelper:
  125 + #auto-dialect: true
  126 + #auto-runtime-dialect: true
  127 + helper-dialect: mysql
  128 + reasonable: true
  129 +
  130 + support-methods-arguments: true
  131 + params: count=countSql
  132 +
  133 +#debug配置,debug或者为true的时候，logback才会记录和写入日志文件
  134 +trace: false
  135 +debug: true
  136 +
  137 +logging:
  138 + file:
  139 + path: ./logs/
  140 + name: system.log
  141 + config: config/logback-dev.xml
  142 + #转移到logback配置文件中
  143 + #level:
  144 + #org.apache.tomcat: info
  145 + #com.tianbo.warehouse.dao: DEBUG
  146 + #org.springframework.security: trace
  147 + #日志配置,输出到文本，
  148 +#Java Web Token 时效时间,单位秒
  149 +jwt:
  150 + max-alive: 30000
  151 +message-bus:
  152 + url:
  153 + #账户登录地址
  154 + login-url: http://36.134.143.171:8083/api/zz-uaa/common/bus/login
  155 + #心跳地址
  156 + hearbit-url: http://36.134.143.171:8083/api/cloud-user-center-heartbeat/heartbeat
  157 + #报文发送地址
  158 + send-url: http://36.134.143.171:8083/api/kafka-server-producer/kafka/send
  159 + #报文接收地址
  160 + get-url: http://36.134.143.171:8083/api/kafka-server-consumer/kafka/receive
  161 + auth:
  162 + username: yangyucheng
  163 + password: 111111
  164 + #心跳间隔时间默认10秒，单位毫秒
  165 + heartbit-interval: 10000
  166 +info:
  167 + version: 1.0
  168 + description: "消息总线-消息转发服务。[转发大数据小组消息到总线上]"
  1 +<?xml version="1.0" encoding="UTF-8"?>
  2 +<!--参考文档链接:https://blog.csdn.net/qq_34912478/article/details/80877132-->
  3 +<!-- 日志级别从低到高分为TRACE < DEBUG < INFO < WARN < ERROR < FATAL，如果设置为WARN，则低于WARN的信息都不会输出 -->
  4 +<!-- scan:当此属性设置为true时，配置文件如果发生改变，将会被重新加载，默认值为true -->
  5 +<!-- scanPeriod:设置监测配置文件是否有修改的时间间隔，如果没有给出时间单位，默认单位是毫秒。当scan为true时，此属性生效。默认的时间间隔为1分钟。 -->
  6 +<!-- debug:当此属性设置为true时，将打印出logback内部日志信息，实时查看logback运行状态。默认值为false。 -->
  7 +<configuration scan="true" scanPeriod="60 seconds">
  8 +
  9 + <!--<include resource="org/springframework/boot/logging/logback/base.xml" />-->
  10 +
  11 + <contextName>logback</contextName>
  12 + <!-- name的值是变量的名称，value的值时变量定义的值。通过定义的值会被插入到logger上下文中。定义变量后，可以使“${}”来使用变量。 -->
  13 + <property name="log.path" value="./logs" />
  14 +
  15 + <!-- 彩色日志 -->
  16 + <!-- 彩色日志依赖的渲染类 -->
  17 + <conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter" />
  18 + <conversionRule conversionWord="wex" converterClass="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter" />
  19 + <conversionRule conversionWord="wEx" converterClass="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter" />
  20 + <!-- 彩色日志格式 -->
  21 + <property name="CONSOLE_LOG_PATTERN" value="${CONSOLE_LOG_PATTERN:-%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}"/>
  22 +
  23 +
  24 + <!--输出到控制台-->
  25 + <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
  26 + <!--此日志appender是为开发使用，只配置最底级别，控制台输出的日志级别是大于或等于此级别的日志信息-->
  27 + <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
  28 + <level>debug</level>
  29 + </filter>
  30 + <encoder>
  31 + <Pattern>${CONSOLE_LOG_PATTERN}</Pattern>
  32 + <!-- 设置字符集 -->
  33 + <charset>UTF-8</charset>
  34 + </encoder>
  35 + </appender>
  36 +
  37 +
  38 + <!--输出到文件-->
  39 +
  40 + <!-- 时间滚动输出 level为 DEBUG 日志 -->
  41 + <appender name="DEBUG_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
  42 + <!-- 正在记录的日志文件的路径及文件名 -->
  43 + <file>${log.path}/log_debug.log</file>
  44 + <!--日志文件输出格式-->
  45 + <encoder>
  46 + <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
  47 + <charset>UTF-8</charset> <!-- 设置字符集 -->
  48 + </encoder>
  49 + <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
  50 + <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
  51 + <!-- 日志归档 -->
  52 + <fileNamePattern>${log.path}/debug/log-debug-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
  53 + <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
  54 + <maxFileSize>100MB</maxFileSize>
  55 + </timeBasedFileNamingAndTriggeringPolicy>
  56 + <!--日志文件保留天数-->
  57 + <maxHistory>15</maxHistory>
  58 + </rollingPolicy>
  59 + <!-- 此日志文件只记录debug级别的 -->
  60 + <filter class="ch.qos.logback.classic.filter.LevelFilter">
  61 + <level>debug</level>
  62 + <onMatch>ACCEPT</onMatch>
  63 + <onMismatch>DENY</onMismatch>
  64 + </filter>
  65 + </appender>
  66 +
  67 + <!-- 时间滚动输出 level为 INFO 日志 -->
  68 + <appender name="INFO_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
  69 + <!-- 正在记录的日志文件的路径及文件名 -->
  70 + <file>${log.path}/log_info.log</file>
  71 + <!--日志文件输出格式-->
  72 + <encoder>
  73 + <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
  74 + <charset>UTF-8</charset>
  75 + </encoder>
  76 + <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
  77 + <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
  78 + <!-- 每天日志归档路径以及格式 -->
  79 + <fileNamePattern>${log.path}/info/log-info-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
  80 + <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
  81 + <maxFileSize>100MB</maxFileSize>
  82 + </timeBasedFileNamingAndTriggeringPolicy>
  83 + <!--日志文件保留天数-->
  84 + <maxHistory>15</maxHistory>
  85 + </rollingPolicy>
  86 + <!-- 此日志文件只记录info级别的 -->
  87 + <filter class="ch.qos.logback.classic.filter.LevelFilter">
  88 + <level>info</level>
  89 + <onMatch>ACCEPT</onMatch>
  90 + <onMismatch>DENY</onMismatch>
  91 + </filter>
  92 + </appender>
  93 +
  94 + <!-- 时间滚动输出 level为 WARN 日志 -->
  95 + <appender name="WARN_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
  96 + <!-- 正在记录的日志文件的路径及文件名 -->
  97 + <file>${log.path}/log_warn.log</file>
  98 + <!--日志文件输出格式-->
  99 + <encoder>
  100 + <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
  101 + <charset>UTF-8</charset> <!-- 此处设置字符集 -->
  102 + </encoder>
  103 + <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
  104 + <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
  105 + <fileNamePattern>${log.path}/warn/log-warn-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
  106 + <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
  107 + <maxFileSize>100MB</maxFileSize>
  108 + </timeBasedFileNamingAndTriggeringPolicy>
  109 + <!--日志文件保留天数-->
  110 + <maxHistory>15</maxHistory>
  111 + </rollingPolicy>
  112 + <!-- 此日志文件只记录warn级别的 -->
  113 + <filter class="ch.qos.logback.classic.filter.LevelFilter">
  114 + <level>warn</level>
  115 + <onMatch>ACCEPT</onMatch>
  116 + <onMismatch>DENY</onMismatch>
  117 + </filter>
  118 + </appender>
  119 +
  120 +
  121 + <!-- 时间滚动输出 level为 ERROR 日志 -->
  122 + <appender name="ERROR_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
  123 + <!-- 正在记录的日志文件的路径及文件名 -->
  124 + <file>${log.path}/log_error.log</file>
  125 + <!--日志文件输出格式-->
  126 + <encoder>
  127 + <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
  128 + <charset>UTF-8</charset> <!-- 此处设置字符集 -->
  129 + </encoder>
  130 + <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
  131 + <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
  132 + <fileNamePattern>${log.path}/error/log-error-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
  133 + <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
  134 + <maxFileSize>100MB</maxFileSize>
  135 + </timeBasedFileNamingAndTriggeringPolicy>
  136 + <!--日志文件保留天数-->
  137 + <maxHistory>15</maxHistory>
  138 + </rollingPolicy>
  139 + <!-- 此日志文件只记录ERROR级别的 -->
  140 + <filter class="ch.qos.logback.classic.filter.LevelFilter">
  141 + <level>ERROR</level>
  142 + <onMatch>ACCEPT</onMatch>
  143 + <onMismatch>DENY</onMismatch>
  144 + </filter>
  145 + </appender>
  146 +
  147 + <!--
  148 + <logger>用来设置某一个包或者具体的某一个类的日志打印级别、
  149 + 以及指定<appender>。<logger>仅有一个name属性，
  150 + 一个可选的level和一个可选的addtivity属性。
  151 + name:用来指定受此logger约束的某一个包或者具体的某一个类。
  152 + level:用来设置打印级别，大小写无关：TRACE, DEBUG, INFO, WARN, ERROR, ALL 和 OFF，
  153 + 还有一个特俗值INHERITED或者同义词NULL，代表强制执行上级的级别。
  154 + 如果未设置此属性，那么当前logger将会继承上级的级别。
  155 + addtivity:是否向上级logger传递打印信息。默认是true。
  156 + -->
  157 + <!--<logger name="org.springframework.web" level="info"/>-->
  158 + <!--<logger name="org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor" level="INFO"/>-->
  159 + <!--
  160 + 使用mybatis的时候，sql语句是debug下才会打印，而这里我们只配置了info，所以想要查看sql语句的话，有以下两种操作：
  161 + 第一种把<root level="info">改成<root level="DEBUG">这样就会打印sql，不过这样日志那边会出现很多其他消息
  162 + 第二种就是单独给dao下目录配置debug模式，代码如下，这样配置sql语句会打印，其他还是正常info级别：
  163 + -->
  164 +
  165 +
  166 + <!--
  167 + root节点是必选节点，用来指定最基础的日志输出级别，只有一个level属性
  168 + level:用来设置打印级别，大小写无关：TRACE, DEBUG, INFO, WARN, ERROR, ALL 和 OFF，
  169 + 不能设置为INHERITED或者同义词NULL。默认是DEBUG
  170 + 可以包含零个或多个元素，标识这个appender将会添加到这个logger。
  171 + -->
  172 + <logger name="com.sunyo.usercenter.heartbeat.dao" level="debug" />
  173 + <!--开发环境:打印控制台-->
  174 + <springProfile name="dev">
  175 + <logger name="org.apache.tomcat" level="info" />
  176 + <logger name="com.sunyo.usercenter.heartbeat" level="debug" />
  177 + <root level="INFO">
  178 + <appender-ref ref="CONSOLE" />
  179 + <appender-ref ref="DEBUG_FILE" />
  180 + <appender-ref ref="INFO_FILE" />
  181 + <appender-ref ref="WARN_FILE" />
  182 + <appender-ref ref="ERROR_FILE" />
  183 + </root>
  184 + </springProfile>
  185 +
  186 + <!--生产环境:输出到文件-->
  187 + <springProfile name="pro">
  188 + <root level="info">
  189 + <appender-ref ref="CONSOLE" />
  190 + <appender-ref ref="DEBUG_FILE" />
  191 + <appender-ref ref="INFO_FILE" />
  192 + <appender-ref ref="ERROR_FILE" />
  193 + <appender-ref ref="WARN_FILE" />
  194 + </root>
  195 + </springProfile>
  196 +
  197 +</configuration>
  1 +<?xml version="1.0" encoding="UTF-8"?>
  2 +<project xmlns="http://maven.apache.org/POM/4.0.0"
  3 + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4 + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5 + <modelVersion>4.0.0</modelVersion>
  6 + <packaging>jar</packaging>
  7 + <groupId>com.tianbo</groupId>
  8 + <artifactId>messagebus-trans-message</artifactId>
  9 + <version>1.0-SNAPSHOT</version>
  10 + <description>消息转发服务</description>
  11 +
  12 + <parent>
  13 + <groupId>org.springframework.boot</groupId>
  14 + <artifactId>spring-boot-starter-parent</artifactId>
  15 + <version>2.2.11.RELEASE</version>
  16 + <relativePath/> <!-- lookup parent from repository -->
  17 + </parent>
  18 +
  19 + <properties>
  20 + <java.version>1.8</java.version>
  21 + <spring-cloud.version>Hoxton.SR9</spring-cloud.version>
  22 + <swagger2_version>2.9.2</swagger2_version>
  23 + <fastjson_version>1.2.73</fastjson_version>
  24 + <lombok_sersion>1.18.6</lombok_sersion>
  25 + </properties>
  26 +
  27 + <dependencies>
  28 + <dependency>
  29 + <groupId>org.springframework.boot</groupId>
  30 + <artifactId>spring-boot-starter-web</artifactId>
  31 + </dependency>
  32 + <dependency>
  33 + <groupId>org.springframework.cloud</groupId>
  34 + <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
  35 + </dependency>
  36 + <!--lombok-->
  37 + <dependency>
  38 + <groupId>org.projectlombok</groupId>
  39 + <artifactId>lombok</artifactId>
  40 + <version>${lombok_sersion}</version>
  41 + <scope>provided</scope>
  42 + </dependency>
  43 + <dependency>
  44 + <groupId>com.alibaba</groupId>
  45 + <artifactId>fastjson</artifactId>
  46 + <version>${fastjson_version}</version>
  47 + </dependency>
  48 + </dependencies>
  49 + <dependencyManagement>
  50 + <dependencies>
  51 + <dependency>
  52 + <groupId>org.springframework.cloud</groupId>
  53 + <artifactId>spring-cloud-dependencies</artifactId>
  54 + <version>${spring-cloud.version}</version>
  55 + <type>pom</type>
  56 + <scope>import</scope>
  57 + </dependency>
  58 + </dependencies>
  59 + </dependencyManagement>
  60 + <build>
  61 + <plugins>
  62 + <plugin>
  63 + <groupId>org.springframework.boot</groupId>
  64 + <artifactId>spring-boot-maven-plugin</artifactId>
  65 + </plugin>
  66 + </plugins>
  67 + </build>
  68 +</project>
  1 +package com.tianbo.messagebus;
  2 +
  3 +import org.springframework.boot.SpringApplication;
  4 +import org.springframework.boot.autoconfigure.SpringBootApplication;
  5 +import org.springframework.boot.web.client.RestTemplateBuilder;
  6 +import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
  7 +import org.springframework.context.annotation.Bean;
  8 +import org.springframework.scheduling.TaskScheduler;
  9 +import org.springframework.scheduling.annotation.EnableScheduling;
  10 +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
  11 +import org.springframework.web.client.RestTemplate;
  12 +
  13 +@SpringBootApplication
  14 +@EnableEurekaClient
  15 +@EnableScheduling
  16 +public class MessageTransApplication {
  17 +
  18 + public static void main(String[] args) {
  19 + SpringApplication.run(MessageTransApplication.class, args);
  20 + }
  21 +
  22 + @Bean
  23 + public RestTemplate restTemplate(RestTemplateBuilder builder) {
  24 + return builder.build();
  25 + }
  26 +
  27 + @Bean
  28 + public TaskScheduler taskScheduler() {
  29 + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
  30 + taskScheduler.setPoolSize(10);
  31 + return taskScheduler;
  32 + }
  33 +
  34 +
  35 +}
  1 +package com.tianbo.messagebus.controller;
  2 +
  3 +import com.alibaba.fastjson.JSON;
  4 +import com.alibaba.fastjson.JSONObject;
  5 +import com.tianbo.messagebus.service.MessageBusProcessor;
  6 +import org.apache.commons.lang.StringUtils;
  7 +import org.springframework.beans.factory.annotation.Autowired;
  8 +import org.springframework.web.bind.annotation.PostMapping;
  9 +import org.springframework.web.bind.annotation.RequestMapping;
  10 +import org.springframework.web.bind.annotation.RestController;
  11 +
  12 +import javax.annotation.Resource;
  13 +import javax.servlet.http.HttpServletRequest;
  14 +import javax.servlet.http.HttpServletResponse;
  15 +import java.io.Serializable;
  16 +import java.util.Date;
  17 +
  18 +@RestController()
  19 +@RequestMapping("/")
  20 +public class HeartbeatController implements Serializable {
  21 + private static final long serialVersionUID = 1L;
  22 +
  23 +
  24 + @Autowired
  25 + private HttpServletRequest request;
  26 +
  27 + @Autowired
  28 + MessageBusProcessor messageBusDemo;
  29 +
  30 + @PostMapping("login")
  31 + public void login(){
  32 + messageBusDemo.login();
  33 + }
  34 +
  35 + @PostMapping("getmsg")
  36 + public void getmsg(){
  37 + messageBusDemo.getMsg();
  38 + }
  39 +
  40 +}
  1 +package com.tianbo.messagebus.controller.response;
  2 +
  3 +import java.io.Serializable;
  4 +
  5 +public class ResultJson<T> implements Serializable{
  6 +
  7 + private static final long serialVersionUID = 1L;
  8 +
  9 + // 状态码 正确为200
  10 + private String code = "200";
  11 + // 描述
  12 + private String msg = "";
  13 +
  14 + private String error;
  15 + // 返回对象
  16 + private T data;
  17 + //返回的JWT
  18 + private String jwtToken;
  19 +
  20 + public ResultJson() {
  21 + }
  22 +
  23 + public ResultJson(String code) {
  24 + this.code = code;
  25 + }
  26 +
  27 + public ResultJson(String code, String msg) {
  28 + this.code = code;
  29 + this.msg = msg;
  30 + }
  31 +
  32 + public ResultJson(T data) {
  33 + this.data = data;
  34 + }
  35 +
  36 + public ResultJson(String code, String msg, T data) {
  37 + this.code = code;
  38 + this.msg = msg;
  39 + this.data = data;
  40 + }
  41 +
  42 + public ResultJson(String code, String msg, T data,String jwtToken) {
  43 + this.code = code;
  44 + this.msg = msg;
  45 + this.data = data;
  46 + this.jwtToken = jwtToken;
  47 + }
  48 +
  49 + public String getCode() {
  50 + return code;
  51 + }
  52 +
  53 + public void setCode(String code) {
  54 + this.code = code;
  55 + }
  56 +
  57 + public String getMsg() {
  58 + return msg;
  59 + }
  60 +
  61 + public void setMsg(String msg) {
  62 + this.msg = msg;
  63 + }
  64 +
  65 + public String getError() {
  66 + return error;
  67 + }
  68 +
  69 + public void setError(String error) {
  70 + this.error = error;
  71 + }
  72 +
  73 + public T getData() {
  74 + return data;
  75 + }
  76 +
  77 + public void setData(T data) {
  78 + this.data = data;
  79 + }
  80 +
  81 + public String getJwtToken() {
  82 + return jwtToken;
  83 + }
  84 +
  85 + public void setJwtToken(String jwtToken) {
  86 + this.jwtToken = jwtToken;
  87 + }
  88 +}
  1 +package com.tianbo.messagebus.service;
  2 +
  3 +import com.alibaba.fastjson.JSON;
  4 +import com.alibaba.fastjson.JSONArray;
  5 +import com.alibaba.fastjson.JSONObject;
  6 +import lombok.extern.slf4j.Slf4j;
  7 +import org.springframework.beans.factory.annotation.Value;
  8 +import org.springframework.http.*;
  9 +import org.springframework.scheduling.annotation.Async;
  10 +import org.springframework.scheduling.annotation.EnableAsync;
  11 +import org.springframework.scheduling.annotation.Scheduled;
  12 +import org.springframework.stereotype.Service;
  13 +import org.springframework.util.LinkedMultiValueMap;
  14 +import org.springframework.util.MultiValueMap;
  15 +import org.springframework.util.StringUtils;
  16 +import org.springframework.web.client.RestTemplate;
  17 +
  18 +import javax.annotation.Resource;
  19 +import java.io.Serializable;
  20 +
  21 +@Service
  22 +@EnableAsync
  23 +@Slf4j
  24 +public class MessageBusProcessor {
  25 +
  26 + /**http://10.161.4.20:8083/
  27 + * 账户登录地址
  28 + */
  29 + @Value("${message-bus.url.login-url}")
  30 + private String LOGIN_URL;
  31 + /**
  32 + * 账号名
  33 + */
  34 + @Value("${message-bus.auth.username}")
  35 + private String USER_NAME;
  36 +// private static final String USER_NAME = "HYYW";
  37 + /**
  38 + * 登陆密码
  39 + */
  40 + @Value("${message-bus.auth.password}")
  41 + private String USER_PASS;
  42 +// private static final String USER_PASS = "ZZecargo123";
  43 + /**
  44 + * 心跳接口地址
  45 + */
  46 + @Value("${message-bus.url.hearbit-url}")
  47 + private String HEARTBEAT_URL;
  48 + /**
  49 + * 心跳间隔时间 单位S
  50 + */
  51 + @Value("${message-bus.heartbit-interval}")
  52 + private int HEARTBIT_INTERVAL;
  53 + /**
  54 + * 发送报文地址
  55 + */
  56 + @Value("${message-bus.url.send-url}")
  57 + private String SEND_MSG_URL;
  58 + /**
  59 + * 接收报文地址
  60 + */
  61 + @Value("${message-bus.url.get-url}")
  62 + private String GET_MSG_URL;
  63 + /**
  64 + * 存储登录后的token
  65 + */
  66 + private static String TOKEN = "";
  67 + /**
  68 + * 登录成功状态
  69 + */
  70 + private static Boolean LOGIN_STATUS=false;
  71 +
  72 +
  73 + /**
  74 + * HTTP请求框架
  75 + */
  76 + @Resource
  77 + private RestTemplate restTemplate;
  78 +
  79 + /**
  80 + * 发起登录，存储token
  81 + *
  82 + * @return
  83 + */
  84 + public Boolean login() {
  85 + try {
  86 + /*
  87 + * 发起HTTP 登录请求
  88 + * 登录接口的请求头为application/x-www-form-urlencoded
  89 + */
  90 + HttpHeaders headers = new HttpHeaders();
  91 + headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
  92 +
  93 + /*
  94 + * 请求参数
  95 + */
  96 + MultiValueMap<String, String> params = new LinkedMultiValueMap<String, String>();
  97 + params.add("username", USER_NAME);
  98 + params.add("password", USER_PASS);
  99 +
  100 + HttpEntity<MultiValueMap<String, String>> request = new HttpEntity<MultiValueMap<String, String>>(params, headers);
  101 +
  102 + /*
  103 + * 提交HTTP访问，获取返回信息
  104 + */
  105 + ResponseEntity<String> response = restTemplate.postForEntity(LOGIN_URL, request, String.class);
  106 + // 输出结果
  107 + log.info(response.getBody());
  108 +
  109 +
  110 + /*
  111 + 校验是否登录成功
  112 + */
  113 + if (response.getStatusCode().equals(HttpStatus.OK)) {
  114 + /**
  115 + * 从返回信息中确定是否登录成功，并取得token
  116 + * 返回格式
  117 + * {
  118 + "code":200,
  119 + "data":{
  120 + "account":"yangyucheng",
  121 + "token":"eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiI2MCIsIm5hbWUiOiLmnajnjonmiJDmtYvor5UiLCJpZCI6NjAsImlhdCI6MTYxNzM0ODM3MiwiYWNjb3VudCI6Inlhbmd5dWNoZW5nIn0.ElAs7BtV1tu6ApQXuPXzgXUgvja76bjEb-zxqhUON48"
  122 + },
  123 + "message":"success",
  124 + "success":true,
  125 + "time":20210402152612604
  126 + }
  127 + */
  128 + JSONObject resJson = JSON.parseObject(response.getBody());
  129 + JSONObject resData = resJson.getJSONObject("data");
  130 + String resCode = resJson.getString("code");
  131 +
  132 +
  133 +
  134 + /*
  135 + 校验并获取登陆成功后返回的token
  136 + */
  137 + String authToken = resData.getString("token");
  138 + if ("200".equals(resCode) && !StringUtils.isEmpty(authToken) && authToken.length() > 10) {
  139 + LOGIN_STATUS = true;
  140 +
  141 + //设置请求头部Authorization为token, token的类型为Bearer
  142 + TOKEN = authToken;
  143 + /*
  144 + 登录成功开始心跳
  145 + */
  146 +// startHeartBit();
  147 + log.info("登录成功");
  148 + return true;
  149 + }else {
  150 + log.error("登录失败");
  151 + return false;
  152 + }
  153 + } else {
  154 + log.error("登录失败");
  155 + return false;
  156 + }
  157 +
  158 + } catch (Exception e) {
  159 + log.error("登录失败->{}",e.toString());
  160 + return false;
  161 + }
  162 + }
  163 +
  164 + /**
  165 + * 定时心跳，维持在线状态，每10秒访问一次心跳接口
  166 + */
  167 + public void startHeartBit() {
  168 + try {
  169 + while (true) {
  170 + heartBit();
  171 + Thread.sleep(HEARTBIT_INTERVAL);
  172 + }
  173 + } catch (InterruptedException e) {
  174 + e.printStackTrace();
  175 + }
  176 + }
  177 +
  178 + @Scheduled(fixedRate = 10000)
  179 + public void heartBit() {
  180 + if (!StringUtils.isEmpty(TOKEN) && LOGIN_STATUS){
  181 + /*
  182 + * 发起HTTP 登录请求
  183 + * 登录接口的请求头为application/x-www-form-urlencoded
  184 + */
  185 + HttpHeaders headers = new HttpHeaders();
  186 + headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
  187 + /*
  188 + * 设置获取到的token到头部信息Authorization节点中
  189 + */
  190 + headers.setBearerAuth(TOKEN);
  191 +
  192 + /*
  193 + * 心跳接口无参数，访问接口即可
  194 + */
  195 + MultiValueMap<String, String> params = new LinkedMultiValueMap<String, String>();
  196 +
  197 + HttpEntity<MultiValueMap<String, String>> request = new HttpEntity<MultiValueMap<String, String>>(params, headers);
  198 +
  199 + /*
  200 + * 提交HTTP访问，获取返回信息
  201 + */
  202 + ResponseEntity<String> response = restTemplate.postForEntity(HEARTBEAT_URL, request, String.class);
  203 + // 输出结果
  204 + System.out.println(response.getBody());
  205 + if (response.getStatusCode().equals(HttpStatus.OK)) {
  206 + log.info("心跳成功");
  207 + } else {
  208 + log.error("心跳失败");
  209 + }
  210 + }
  211 + }
  212 +
  213 + /**
  214 + * 发送消息
  215 + *
  216 + * @return
  217 + */
  218 + public Boolean sendMsg(MSG msg) {
  219 + if (LOGIN_STATUS) {
  220 + /*
  221 + * 发起HTTP 登录请求
  222 + * 登录接口的请求头为application/json
  223 + */
  224 + HttpHeaders headers = new HttpHeaders();
  225 + headers.setContentType(MediaType.APPLICATION_JSON);
  226 +
  227 + MSGS msgs = new MSGS();
  228 + msg.getHEADER().setSNDR("HYYW");
  229 +
  230 + msgs.setMSG(msg);
  231 +
  232 +
  233 + /*
  234 + 设置要发送的实体类并将实体转换成Json字符串,这里以MAP类型举例
  235 + */
  236 +// Map<String, String> dataModel = new HashMap<>();
  237 +// dataModel.put("flightNo", "CV987");
  238 +// dataModel.put("flightDate", "MAY01");
  239 +// dataModel.put("waybillNo", "172-12345678");
  240 +// dataModel.put("weight", "20.01");
  241 +// dataModel.put("piece", "2");
  242 +// msg.getMSG().setBODY(JSON.toJSONString(dataModel));
  243 +
  244 + /*
  245 + * 设置获取到的token到头部信息Authorization节点中
  246 + */
  247 + headers.setBearerAuth(TOKEN);
  248 +
  249 + /*
  250 + * 发起消息接口访问，发送消息
  251 + */
  252 +
  253 + HttpEntity<MSGS> request = new HttpEntity<MSGS>(msgs, headers);
  254 + ResponseEntity<String> response = restTemplate.postForEntity(SEND_MSG_URL, request, String.class);
  255 +
  256 + JSONObject resJson = JSON.parseObject(response.getBody());
  257 + String code = resJson.getString("code");
  258 +
  259 + System.out.println(response.getBody());
  260 +
  261 + if (response.getStatusCode().equals(HttpStatus.OK) && "200".equals(code)) {
  262 + log.info("消息发送成功");
  263 + return true;
  264 + } else {
  265 + log.error("消息发送失败");
  266 + return false;
  267 + }
  268 +
  269 + }
  270 + return false;
  271 + }
  272 +
  273 +
  274 + /**
  275 + * 收取消息
  276 + *
  277 + * @return
  278 + */
  279 + @Async
  280 + @Scheduled(fixedRate = 300)
  281 + public JSONArray getMsg() {
  282 + if(!LOGIN_STATUS){
  283 + login();
  284 + return null;
  285 + }
  286 + /*
  287 + * 发起HTTP 登录请求
  288 + * 登录接口的请求头为application/json
  289 + */
  290 + HttpHeaders headers = new HttpHeaders();
  291 + headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
  292 + headers.setBearerAuth(TOKEN);
  293 + /*
  294 + * 请求参数拼装
  295 + */
  296 + MultiValueMap<String, String> params = new LinkedMultiValueMap<String, String>();
  297 + params.add("username", "HYYW");
  298 + HttpEntity<MultiValueMap<String, String>> request = new HttpEntity<MultiValueMap<String, String>>(params, headers);
  299 +
  300 + /*
  301 + * 提交HTTP访问，获取返回信息
  302 + */
  303 + ResponseEntity<String> response = restTemplate.postForEntity(GET_MSG_URL, request, String.class);
  304 + // 输出结果
  305 + log.info("获取到消息返回---{}---",response.getBody());
  306 + /*
  307 + * 从返回信息中确定是否获取到消息
  308 + */
  309 + JSONObject resJson = JSON.parseObject(response.getBody());
  310 + String code = resJson.getString("code");
  311 + if (response.getStatusCode().equals(HttpStatus.OK) && "200".equals(code)) {
  312 +
  313 + JSONArray data = resJson.getJSONArray("data");
  314 + log.info("消息接收成功，接收消息为>>>{}<<<",data.toString());
  315 +
  316 + for (int i = 0; i<data.size() ; i++) {
  317 + /*
  318 + 取得是大数据小组的实体,他们的msg.body的封装是以对象实体object封装的。不是json字符串。
  319 + */
  320 + String msg = data.getObject(i,String.class);
  321 + JSONObject rootJson = JSON.parseObject(msg);
  322 + JSONObject msgJson = rootJson.getJSONObject("MSG");
  323 + JSONObject body = msgJson.getJSONObject("BODY");
  324 +
  325 + HEADER msgHeader = msgJson.getObject("HEADER",HEADER.class);
  326 +
  327 + MSG transMsg= new MSG();
  328 + String transBody = body.toJSONString();
  329 + transMsg.setHEADER(msgHeader);
  330 + transMsg.setBODY(transBody);
  331 +
  332 + transMsg.toString();
  333 +
  334 + /*
  335 + 自定义对返回数据的处理
  336 + */
  337 + log.info("开始转发消息");
  338 + Boolean sendResult = sendMsg(transMsg);
  339 + /**
  340 + * todo:消息失败处理
  341 + */
  342 + if(!sendResult){
  343 + //todo:消息备份或者重发？
  344 + }
  345 + }
  346 +
  347 +
  348 +
  349 +
  350 + return data;
  351 +
  352 + } else {
  353 + log.error("消息获取失败");
  354 + return new JSONArray();
  355 + }
  356 + }
  357 +
  358 + /**
  359 + * 读取备份消息并消息重发
  360 + * @return
  361 + */
  362 + public Boolean reSend(){
  363 + return false;
  364 + }
  365 +
  366 +}
  367 +
  368 +/**
  369 + * 消息发送实体类
  370 + */
  371 +class MSGS implements Serializable {
  372 + private MSG MSG;
  373 +
  374 + public MSG getMSG() {
  375 + return MSG;
  376 + }
  377 +
  378 + public void setMSG(MSG MSG) {
  379 + this.MSG = MSG;
  380 + }
  381 +}
  382 +
  383 +class MSG {
  384 + /**
  385 + * 具体消息头部信息
  386 + */
  387 + private HEADER HEADER;
  388 + /**
  389 + * 具体消息支持JSON字符串或者XML
  390 + */
  391 + private String BODY;
  392 +
  393 + public HEADER getHEADER() {
  394 + return HEADER;
  395 + }
  396 +
  397 + public void setHEADER(HEADER HEADER) {
  398 + this.HEADER = HEADER;
  399 + }
  400 +
  401 + public String getBODY() {
  402 + return BODY;
  403 + }
  404 +
  405 + public void setBODY(String BODY) {
  406 + this.BODY = BODY;
  407 + }
  408 +}
  409 +
  410 +/**
  411 + * 消息发送头部信息
  412 + */
  413 +class HEADER {
  414 + /**
  415 + * 消息唯一标识，可以是ID
  416 + */
  417 + private long SEQNO;
  418 + /**
  419 + * 发送的消息大类,在总线系统配置好后，分配给消息生产者
  420 + */
  421 + private String TYPE;
  422 + /**
  423 + * 发送的消息子类，在总线系统配置好后，分配给消息生产者
  424 + */
  425 + private String STYPE;
  426 + /**
  427 + * 消息增删改类型(IE=insert event,UE=update event,DE=delete event)
  428 + */
  429 + private String OPTYPE;
  430 + /**
  431 + * 发送时间(如：20210320101421/yyyyMMddHHmmss)
  432 + */
  433 + private String DDTM;
  434 + /**
  435 + * 你的账号名称
  436 + */
  437 + private String SNDR;
  438 + /**
  439 + * 消息接收者
  440 + */
  441 + private String RCVR;
  442 +
  443 + public long getSEQNO() {
  444 + return SEQNO;
  445 + }
  446 +
  447 + public void setSEQNO(long SEQNO) {
  448 + this.SEQNO = SEQNO;
  449 + }
  450 +
  451 + public String getTYPE() {
  452 + return TYPE;
  453 + }
  454 +
  455 + public void setTYPE(String TYPE) {
  456 + this.TYPE = TYPE;
  457 + }
  458 +
  459 + public String getSTYPE() {
  460 + return STYPE;
  461 + }
  462 +
  463 + public void setSTYPE(String STYPE) {
  464 + this.STYPE = STYPE;
  465 + }
  466 +
  467 + public String getOPTYPE() {
  468 + return OPTYPE;
  469 + }
  470 +
  471 + public void setOPTYPE(String OPTYPE) {
  472 + this.OPTYPE = OPTYPE;
  473 + }
  474 +
  475 + public String getDDTM() {
  476 + return DDTM;
  477 + }
  478 +
  479 + public void setDDTM(String DDTM) {
  480 + this.DDTM = DDTM;
  481 + }
  482 +
  483 + public String getSNDR() {
  484 + return SNDR;
  485 + }
  486 +
  487 + public void setSNDR(String SNDR) {
  488 + this.SNDR = SNDR;
  489 + }
  490 +
  491 + public String getRCVR() {
  492 + return RCVR;
  493 + }
  494 +
  495 + public void setRCVR(String RCVR) {
  496 + this.RCVR = RCVR;
  497 + }
  498 +}
  499 +
  500 +/**
  501 + * 收发送接口返回结果实体类
  502 + */
  503 +
  504 +class ResultJson<T> implements Serializable {
  505 + private static final long serialVersionUID = 1L;
  506 +
  507 + /**
  508 + * 状态码 正确为200，其他为异常
  509 + */
  510 + private String code;
  511 + /**
  512 + * 结果描述
  513 + */
  514 + private String msg;
  515 +
  516 + /**
  517 + * 结果数据
  518 + */
  519 + private T data;
  520 +}