Skip to content

向汉 / uc-consumer

Go to a project
Toggle navigation
Toggle navigation pinning
Commit e20d47d7 e20d47d7f019921dbd8202b13e21ebe2ea93ba7b by xianghan
Options

1.优化

1 parent 8c7571ef
Expand all
Inline Side-by-side
Showing 20 changed files with 1032 additions and 404 deletions
......@@ -23,12 +23,12 @@
<dependencies>
<!--redisson-->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.16.3</version>
</dependency>
<!--wechat-util-->
<!--<dependency>
<groupId>com.topdraw</groupId>
<artifactId>wechat-util</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>-->
<!--代码生成器-->
<dependency>
......
package com.topdraw.config;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.naming.ConfigurationException;
import java.util.List;
import java.util.Map;
@Component
public class RabbitMqBindingConfig {
/**************************************************数据源选择*************************************************************/
@Autowired
private RabbitMqSourceConfig rabbitMqSourceConfig;
@Autowired
private RabbitMqCustomConfig rabbitMqCustomConfig;
@Resource(name = "managementRabbitAdmin")
private RabbitAdmin managementRabbitAdmin;
@Resource(name = "serviceRabbitAdmin")
private RabbitAdmin serviceRabbitAdmin;
@PostConstruct
public void initBinding() throws ConfigurationException {
// String source = rabbitMqSourceConfig.getActiveSource();
List<Map<String, String>> list = rabbitMqCustomConfig.getList();
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String exchange = map.get("exchange");
String exchangeType = map.get("exchange-type");
Exchange exchange_ = null;
switch (exchangeType) {
case ExchangeTypes.TOPIC:
exchange_ = ExchangeBuilder.topicExchange(exchange)
.durable(true).build();
break;
case ExchangeTypes.FANOUT:
exchange_ = ExchangeBuilder.fanoutExchange(exchange)
.durable(true).build();
break;
case ExchangeTypes.HEADERS:
exchange_ = ExchangeBuilder.headersExchange(exchange)
.durable(true).build();
break;
default:
exchange_ = ExchangeBuilder.directExchange(exchange)
.durable(true).build();
break;
}
String queue = map.get("queue");
Queue queue_ = new Queue(queue);
String routingKey = map.get("routing-key");
if (StringUtils.isBlank(routingKey)) {
routingKey = queue;
}
Binding binding_ = BindingBuilder.bind(queue_).to(exchange_).with(routingKey).and(null);
String active = map.get("active");
switch (active) {
case "management":
this.managementRabbitAdmin.declareExchange(exchange_);
this.managementRabbitAdmin.declareQueue(queue_);
this.managementRabbitAdmin.declareBinding(binding_);
break;
case "service":
this.serviceRabbitAdmin.declareExchange(exchange_);
this.serviceRabbitAdmin.declareQueue(queue_);
this.serviceRabbitAdmin.declareBinding(binding_);
break;
/* case "service,management":
this.serviceRabbitAdmin.declareExchange(exchange_);
this.serviceRabbitAdmin.declareQueue(queue_);
this.serviceRabbitAdmin.declareBinding(binding_);
this.managementRabbitAdmin.declareExchange(exchange_);
this.managementRabbitAdmin.declareQueue(queue_);
this.managementRabbitAdmin.declareBinding(binding_);
break;*/
default:
break;
}
}
}
}
}
package com.topdraw.config;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
@Configuration
public class RabbitMqConfig {
@Value("${mutil-mq.service.host}")
private String serviceHost;
@Value("${mutil-mq.service.port}")
private Integer servicePort;
@Value("${mutil-mq.service.username}")
private String serviceUserName;
@Value("${mutil-mq.service.password}")
private String servicePassword;
@Value("${mutil-mq.service.virtual-host}")
private String serviceVirtualHost;
@Bean(name = "serviceConnectionFactory")
@Primary
public ConnectionFactory serviceConnectionFactory(){
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost(serviceHost);
connectionFactory.setPort(servicePort);
connectionFactory.setUsername(serviceUserName);
connectionFactory.setPassword(servicePassword);
connectionFactory.setVirtualHost(serviceVirtualHost);
return connectionFactory;
}
@Bean(name = "serviceRabbitListenerContainerFactory")
@Primary
public RabbitListenerContainerFactory userCenterRabbitListenerContainerFactory(
SimpleRabbitListenerContainerFactoryConfigurer containerFactoryConfigurer,
@Qualifier("serviceConnectionFactory") ConnectionFactory connectionFactory) {
System.out.println("userCenterRabbitListenerContainerFactory ====>> start ");
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
containerFactoryConfigurer.configure(factory,connectionFactory);
return factory;
}
@Bean(name = "serviceRabbitTemplate")
public RabbitTemplate userCenterRabbitTemplate(ConnectionFactory ucServiceConnectionFactory){
RabbitTemplate u = new RabbitTemplate(ucServiceConnectionFactory);
return u;
}
@Value("${mutil-mq.management.host}")
private String managementHost;
@Value("${mutil-mq.management.port}")
private Integer managementPort;
@Value("${mutil-mq.management.username}")
private String managementUserName;
@Value("${mutil-mq.management.password}")
private String managementPassword;
@Value("${mutil-mq.management.virtual-host}")
private String managementVirtualHost;
@Bean(name = "managementConnectionFactory")
public ConnectionFactory managementConnectionFactory(){
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost(managementHost);
connectionFactory.setPort(managementPort);
connectionFactory.setUsername(managementUserName);
connectionFactory.setPassword(managementPassword);
connectionFactory.setVirtualHost(managementVirtualHost);
return connectionFactory;
}
@Bean(name = "managementRabbitListenerContainerFactory")
public RabbitListenerContainerFactory memberServiceRabbitListenerContainerFactory(
SimpleRabbitListenerContainerFactoryConfigurer containerFactoryConfigurer,
@Qualifier("managementConnectionFactory") ConnectionFactory connectionFactory) {
System.out.println("memberServiceRabbitListenerContainerFactory ====>> start ");
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
containerFactoryConfigurer.configure(factory,connectionFactory);
return factory;
}
@Bean(name = "managementRabbitTemplate")
public RabbitTemplate memberServiceRabbitTemplate(ConnectionFactory ucGatewayConnectionFactory){
RabbitTemplate u = new RabbitTemplate(ucGatewayConnectionFactory);
return u;
}
/**************************************************uc-getaway-samll 2 uc-consumer-samll*************************************************************/
/*public static final String GATEWAY_TO_SMALL_DIRECT = "uc.gateway.small.direct";
@Bean
public Queue eventSmallDirect() {
return new Queue(GATEWAY_TO_SMALL_DIRECT);
}
@Bean
Binding eventSmallBinding(DirectExchange managementExchange , Queue eventSmallDirect) {
BindingBuilder.DirectExchangeRoutingKeyConfigurer directExchangeRoutingKeyConfigurer =
BindingBuilder.bind(eventSmallDirect).to(managementExchange);
return directExchangeRoutingKeyConfigurer.with(GATEWAY_TO_SMALL_DIRECT);
}*/
/**************************************************uc-engine-management 2 uc-consumer-iptv*************************************************************/
/**************************************************uc-getaway-iptv 2 uc-consumer-iptv*************************************************************/
public static final String GATEWAY_TO_SERVICE_DIRECT = "uc.gateway.service.direct";
@Bean
public Queue eventServiceDirect() {
return new Queue(GATEWAY_TO_SERVICE_DIRECT);
}
@Bean
Binding eventBinding(DirectExchange serviceExchange, Queue eventServiceDirect) {
BindingBuilder.DirectExchangeRoutingKeyConfigurer directExchangeRoutingKeyConfigurer =
BindingBuilder.bind(eventServiceDirect).to(serviceExchange);
return directExchangeRoutingKeyConfigurer.with(ENGINE_TO_SERVICE_DIRECT);
}
/**************************************************uc-engine-management 2 uc-consumer-iptv*************************************************************/
public static final String UC_DIRECT_MANAGEMENT = "uc.direct.management";
// uc-service-management 2 uc-consumer-iptv
// public static final String ENGINE_TO_SERVICE_DIRECT = "uc.engine.service.direct";
public static final String ENGINE_TO_SERVICE_DIRECT = "uc.engine.service.direct";
@Bean
public Queue managementDirect() {
return new Queue(ENGINE_TO_SERVICE_DIRECT);
}
@Bean
DirectExchange managementExchange(){
return ExchangeBuilder.directExchange(UC_DIRECT_MANAGEMENT).build();
}
@Bean
Binding managementBinding(DirectExchange managementExchange , Queue managementDirect) {
BindingBuilder.DirectExchangeRoutingKeyConfigurer directExchangeRoutingKeyConfigurer =
BindingBuilder.bind(managementDirect).to(managementExchange);
return directExchangeRoutingKeyConfigurer.with(ENGINE_TO_SERVICE_DIRECT);
}
/**************************************************uc-engine-iptv 2 uc-consumer-management*************************************************************/
public static final String UC_DIRECT_SERVICE = "uc.direct";
// uc-service-iptv 2 uc-consumer-management
public static final String ENGINE_TO_MANAGEMENT_DIRECT = "uc.engine.management.direct";
@Bean
public Queue serviceDirect() {
return new Queue(ENGINE_TO_MANAGEMENT_DIRECT);
}
@Bean
DirectExchange serviceExchange(){
return ExchangeBuilder.directExchange(UC_DIRECT_SERVICE).build();
}
@Bean
Binding serviceBinding(DirectExchange serviceExchange , Queue serviceDirect) {
BindingBuilder.DirectExchangeRoutingKeyConfigurer directExchangeRoutingKeyConfigurer =
BindingBuilder.bind(serviceDirect).to(serviceExchange);
return directExchangeRoutingKeyConfigurer.with(ENGINE_TO_MANAGEMENT_DIRECT);
}
/**************************************************数据基座*************************************************************/
public static final String UC_EVENTBUS_EXCHANGE = "uc.eventbus";
public static final String UC_EVENTBUS_KEY = "uc.eventbus.*.topic";
public static final String UC_EVENTBUS_QUEUE = "uc.eventbus";
@Bean
public Queue eventBusQueue() {
return new Queue(UC_EVENTBUS_QUEUE);
}
@Bean
TopicExchange eventBusExchange() {
return ExchangeBuilder.topicExchange(UC_EVENTBUS_EXCHANGE)
.durable(true).build();
}
@Bean
Binding eventBusBinding(TopicExchange eventBusExchange, Queue eventBusQueue) {
return BindingBuilder
.bind(eventBusQueue)
.to(eventBusExchange)
.with(UC_EVENTBUS_KEY);
}
/**************************************************跨屏数据*************************************************************/
/** 删除全部收藏队列(大屏->小屏) */
public static final String COLLECTION_DELETE_ALL_QUEUE = "queue.collection.deleteall";
/** 添加收藏队列(大屏->小屏) */
public static final String COLLECTION_ADD_QUEUE = "queue.collection.add";
/** 删除收藏队列(大屏->小屏) */
public static final String COLLECTION_DELETE_QUEUE = "queue.collection.delete";
/** 微信侧 公众号关注与取消关注 */
public static final String WEIXIN_SUBORUNSUB_QUEUE = "weixin.subOrUnSub.queue";
}
package com.topdraw.config;
import lombok.Data;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import java.util.List;
import java.util.Map;
@Data
@Configuration
@ConfigurationProperties(prefix = "service.mq")
public class RabbitMqCustomConfig {
private List<Map<String, String>> list;
/**
* viewRecord
* @return
*/
public Map<String, String> getViewRecordInfo() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("source");
if (type.equalsIgnoreCase("viewRecord")) {
return map;
}
}
}
return null;
}
/**
* ucg_event
* @return
*/
public Map<String, String> getUcgEventInfo() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("source");
if (type.equalsIgnoreCase("event")) {
return map;
}
}
}
return null;
}
/**
* ucg_collection
* @return
*/
public Map<String, String> getUcgCollectionInfo() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("source");
if (type.equalsIgnoreCase("collection")) {
return map;
}
}
}
return null;
}
/**
* uce
* @return
*/
public Map<String, String> getUceInfo() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("source");
if (type.equalsIgnoreCase("uce")) {
return map;
}
}
}
return null;
}
/**
* eventBus
* @return
*/
public Map<String, String> getEventBusInfo() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("source");
if (type.equalsIgnoreCase("eventBus")) {
return map;
}
}
}
return null;
}
/**
* wechat
* @return
*/
public Map<String, String> getWechatInfo() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("source");
if (type.equalsIgnoreCase("wechat")) {
return map;
}
}
}
return null;
}
}
package com.topdraw.config;
import lombok.Data;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import java.util.List;
import java.util.Map;
/**
* @author ：
* @description：
* @function :
* @date ：Created in 2022/5/12 12:37
* @version: ：
* @modified By：
* @since : modified in 2022/5/12 12:37
*/
@Data
@Configuration
@ConfigurationProperties(prefix = "service.mq.error.logs")
public class RabbitMqErrorLogConfig {
private List<Map<String, String>> list;
/**
* uce
* @return
*/
public Map<String, String> getWechatError() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("type");
if (type.equalsIgnoreCase("wechat")) {
return map;
}
}
}
return null;
}
/**
* uce
* @return
*/
public Map<String, String> getUceError() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("type");
if (type.equalsIgnoreCase("uce")) {
return map;
}
}
}
return null;
}
/**
* ucg
* @return
*/
public Map<String, String> getUcgError() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("type");
if (type.equalsIgnoreCase("ucg")) {
return map;
}
}
}
return null;
}
/**
* 数据总线
* @return
*/
public Map<String, String> getEventBusError() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("type");
if (type.equalsIgnoreCase("eventBus")) {
return map;
}
}
}
return null;
}
/**
* 个人信息完善
* @return
*/
public Map<String, String> getCompleteInfoError() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("type");
if (type.equalsIgnoreCase("completeInfo")) {
return map;
}
}
}
return null;
}
/**
* 订购
* @return
*/
public Map<String, String> getOrderError() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("type");
if (type.equalsIgnoreCase("order")) {
return map;
}
}
}
return null;
}
/**
* 参加活动
* @return
*/
public Map<String, String> getActivityError() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("type");
if (type.equalsIgnoreCase("activity")) {
return map;
}
}
}
return null;
}
/**
* 签到
* @return
*/
public Map<String, String> getSignError() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("type");
if (type.equalsIgnoreCase("sign")) {
return map;
}
}
}
return null;
}
/**
* 观影、播放记录
* @return
*/
public Map<String, String> getPlayError() {
if (CollectionUtils.isNotEmpty(list)) {
for (Map<String, String> map : list) {
String type = map.get("type");
if (type.equalsIgnoreCase("play")) {
return map;
}
}
}
return null;
}
}
package com.topdraw.exception;
/**
* @author ：
* @description：
* @function :
* @date ：Created in 2022/5/13 15:50
* @version: ：
* @modified By：
* @since : modified in 2022/5/13 15:50
*/
public class RabbitMqExceptionHandler {
}
package com.topdraw.mq.consumer;
import com.topdraw.config.RabbitMqConfig;
import com.topdraw.mq.domain.DataSyncMsg;
import com.rabbitmq.client.Channel;
import com.topdraw.mq.domain.TableOperationMsg;
import com.topdraw.resttemplate.RestTemplateClient;
import com.topdraw.util.FileUtil;
import com.topdraw.util.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.apache.commons.collections4.MapUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import java.io.IOException;
import java.time.LocalDate;
import java.util.Map;
@Component
@Slf4j
public class UcEngineIptv2ManagementConsumer {
......@@ -24,6 +29,9 @@ public class UcEngineIptv2ManagementConsumer {
@Autowired
RestTemplateClient restTemplateClient;
@Value("#{rabbitMqErrorLogConfig.getUceError()}")
private Map<String, String> error;
/**
* 事件
* @param content
......@@ -32,14 +40,36 @@ public class UcEngineIptv2ManagementConsumer {
* @date 2021/9/7 11:26 上午
*/
/*@RabbitHandler
@RabbitListener(bindings = {
@QueueBinding(value = @Queue(value = RabbitMqConfig.ENGINE_TO_MANAGEMENT_DIRECT),
exchange = @Exchange(value = ExchangeTypes.DIRECT))
}, containerFactory = "serviceRabbitListenerContainerFactory")*/
public void ucEventConsumer(String content) {
@RabbitListener(queues = "#{rabbitMqConfig.getUceQueue()}",
containerFactory = "serviceRabbitListenerContainerFactory", ackMode = "MANUAL")*/
public void ucEventConsumer(Channel channel, Message message, String content) throws IOException {
log.info(" receive dataSync msg , content is : {} ", content);
try {
TableOperationMsg tableOperationMsg = this.parseContent(content);
autoUser.route(tableOperationMsg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}
e.printStackTrace();
}
log.info("ucEventConsumer ====>>>> end");
}
......
package com.topdraw.mq.consumer;
import com.topdraw.config.RabbitMqConfig;
import com.rabbitmq.client.Channel;
import com.topdraw.mq.domain.TableOperationMsg;
import com.topdraw.resttemplate.RestTemplateClient;
import com.topdraw.util.FileUtil;
import com.topdraw.util.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.apache.commons.collections4.MapUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import java.io.IOException;
import java.time.LocalDate;
import java.util.Map;
@Component
@Slf4j
public class UcEngineManagement2IptvConsumer {
......@@ -21,6 +28,9 @@ public class UcEngineManagement2IptvConsumer {
@Autowired
RestTemplateClient restTemplateClient;
@Value("#{rabbitMqErrorLogConfig.getUceError()}")
private Map<String, String> error;
/**
* 事件
* @param content
......@@ -29,14 +39,38 @@ public class UcEngineManagement2IptvConsumer {
* @date 2021/9/7 11:26 上午
*/
@RabbitHandler
@RabbitListener(bindings = {
@QueueBinding(value = @Queue(value = RabbitMqConfig.ENGINE_TO_SERVICE_DIRECT),
exchange = @Exchange(value = ExchangeTypes.DIRECT))
}, containerFactory = "managementRabbitListenerContainerFactory")
public void ucEventConsumer(String content) {
@RabbitListener(queues = "#{rabbitMqSourceConfig.getUceQueue()}",
containerFactory = "#{rabbitMqSourceConfig.getUceSource()}",
autoStartup = "#{rabbitMqSourceConfig.getUceStartUp()}",
ackMode = "MANUAL")
public void ucEventConsumer(Channel channel, Message message, String content) throws IOException {
log.info(" receive dataSync msg , content is : {} ", content);
try {
TableOperationMsg tableOperationMsg = this.parseContent(content);
autoUser.route(tableOperationMsg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}
e.printStackTrace();
}
log.info("ucEventConsumer ====>>>> end");
}
......
package com.topdraw.mq.consumer;
import com.topdraw.config.RabbitMqConfig;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.topdraw.mq.domain.DataSyncMsg;
import com.topdraw.resttemplate.RestTemplateClient;
import com.topdraw.util.FileUtil;
import com.topdraw.util.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import java.io.IOException;
import java.time.LocalDate;
import java.util.Map;
import java.util.Objects;
@Component
@Slf4j
public class UcGatewayIptv2IptvConsumer {
......@@ -21,6 +32,9 @@ public class UcGatewayIptv2IptvConsumer {
@Autowired
AutoRoute autoUser;
@Value("#{rabbitMqErrorLogConfig.getUcgError()}")
private Map<String, String> error;
/**
* 事件
* @param content
......@@ -29,14 +43,37 @@ public class UcGatewayIptv2IptvConsumer {
* @date 2021/9/7 11:26 上午
*/
@RabbitHandler
@RabbitListener(bindings = {
@QueueBinding(value = @Queue(value = RabbitMqConfig.GATEWAY_TO_SERVICE_DIRECT),
exchange = @Exchange(value = ExchangeTypes.DIRECT))
}, containerFactory = "serviceRabbitListenerContainerFactory")
public void ucEventConsumer(String content) {
@RabbitListener(queues = "#{rabbitMqSourceConfig.getUcgEventQueue()}",
containerFactory = "#{rabbitMqSourceConfig.getUcgEventSource()}",
autoStartup = "#{rabbitMqSourceConfig.getUcgEventStartUp()}",
ackMode = "MANUAL")
public void eventConsumer(Channel channel, Message message, String content) throws IOException {
log.info(" receive dataSync msg , content is : {} ", content);
try {
DataSyncMsg dataSyncMsg = this.parseContent(content);
this.taskDeal(dataSyncMsg);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (StringUtils.isEmpty(errorStart) || errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}
e.printStackTrace();
}
log.info("ucEventConsumer ====>>>> end");
}
......@@ -60,4 +97,115 @@ public class UcGatewayIptv2IptvConsumer {
private void taskDeal(DataSyncMsg dataSyncMsg) {
this.restTemplateClient.dealTask(dataSyncMsg);
}
/**
* @description 添加收藏记录
* @param content 消息内容
*/
@RabbitHandler
@RabbitListener(queues = "#{rabbitMqSourceConfig.getUcgCollectionQueue()}",
containerFactory = "#{rabbitMqSourceConfig.getUcgCollectionSource()}",
autoStartup = "#{rabbitMqSourceConfig.getUcgCollectionStartUp()}",
ackMode = "MANUAL")
public void collectionConsumer(Channel channel, Message message, String content) throws IOException {
log.info("receive UserCollection add message, content {}", content);
try {
JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
if (Objects.nonNull(content)) {
String evt = jsonObject.get("evt").toString();
String msgData = jsonObject.get("msgData").toString();
switch (evt.toUpperCase()) {
// 添加收藏
case "ADDCOLLECTION":
this.restTemplateClient.addCollection(msgData);
break;
// 删除收藏
case "DELETECOLLECTION":
this.restTemplateClient.deleteCollection(msgData);
break;
// 删除全部收藏
case "DELETEALLCOLLECTION":
this.restTemplateClient.deleteAllCollection(msgData);
break;
default:
break;
}
}
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}
e.printStackTrace();
}
}
/**
* @description 处理观影记录
* @param content 消息内容
*/
@RabbitHandler
@RabbitListener(queues = "#{rabbitMqSourceConfig.getViewRecordQueue()}",
containerFactory = "#{rabbitMqSourceConfig.getViewRecordSource()}",
autoStartup = "#{rabbitMqSourceConfig.getViewRecordStartUp()}",
ackMode = "MANUAL")
public void viewRecordConsumer(Channel channel, Message message, String content) throws IOException {
log.info("receive ViewRecord add message, content {}", content);
try {
JSONObject jsonObject = JSON.parseObject(content, JSONObject.class);
if (Objects.nonNull(content)) {
String evt = jsonObject.get("evt").toString();
String msgData = jsonObject.get("msgData").toString();
switch (evt.toUpperCase()) {
// 添加收藏
case "VIEWRECORD":
this.restTemplateClient.dealViewRecord(msgData);
break;
default:
break;
}
}
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+ LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}
e.printStackTrace();
}
}
}
......
......@@ -2,20 +2,23 @@ package com.topdraw.mq.consumer;
import com.alibaba.fastjson.JSONObject;
import com.topdraw.config.RabbitMqConfig;
import com.rabbitmq.client.Channel;
import com.topdraw.mq.domain.SubscribeBean;
import com.topdraw.resttemplate.RestTemplateClient;
import com.topdraw.utils.RedisUtils;
import com.topdraw.utils.StringUtils;
import com.topdraw.util.FileUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.apache.commons.collections4.MapUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import java.io.IOException;
import java.time.LocalDate;
import java.util.Map;
/**
* 微信事件
*/
......@@ -28,42 +31,9 @@ public class WeiXinEventConsumer {
private static final String QR_CODE_URL = "QR_CODE_URL_";
/**
* @description 删除用户收藏记录
* @param content 消息内容
*/
/*@RabbitHandler
@RabbitListener(bindings = {
@QueueBinding(value = @Queue(value = RabbitMqConfig.COLLECTION_DELETE_QUEUE),
exchange = @Exchange(value = ExchangeTypes.DIRECT))},
containerFactory = "managementRabbitListenerContainerFactory")*/
public void deleteCollection(String content) {
try {
log.info("receive UserCollection delete message, content {}", content);
this.restTemplateClient.deleteCollection(content);
} catch (Exception e) {
log.error("CollectionDeleteConsumer || UserCollection delete error || {}", e.toString(), e);
}
}
@Value("#{rabbitMqErrorLogConfig.getWechatError()}")
private Map<String, String> error;
/**
* @description 删除全部收藏记录
* @param content 消息内容
*/
/*@RabbitHandler
@RabbitListener(bindings = {
@QueueBinding(value = @Queue(value = RabbitMqConfig.COLLECTION_DELETE_ALL_QUEUE),
exchange = @Exchange(value = ExchangeTypes.DIRECT))},
containerFactory = "managementRabbitListenerContainerFactory")*/
@Transactional
public void deleteAllCollection(String content) {
try {
log.info("receive UserCollection delete all message, content {}", content);
this.restTemplateClient.deleteAllCollection(content);
} catch (Exception e) {
log.error("CollectionDeleteConsumer || UserCollection delete all error || {}", e.toString(), e);
}
}
/**
* 关注和取关事件
......@@ -74,13 +44,12 @@ public class WeiXinEventConsumer {
* }
* @param content
*/
/* @RabbitHandler
@RabbitListener(bindings = {
@QueueBinding(value = @Queue(value = RabbitMqConfig.WEIXIN_SUBORUNSUB_QUEUE),
exchange = @Exchange(value = ExchangeTypes.DIRECT))},
containerFactory = "managementRabbitListenerContainerFactory")*/
@RabbitHandler
@RabbitListener(queues = "#{rabbitMqSourceConfig.getWechatQueue()}",
containerFactory = "#{rabbitMqSourceConfig.getWechatSource()}",
autoStartup = "#{rabbitMqSourceConfig.getWechatStartUp()}", ackMode = "MANUAL")
@Transactional
public void subOrUnSubEvent(String content) {
public void subOrUnSubEvent(Channel channel, Message message, String content) throws IOException {
try {
log.info("receive wxu subOrUnSub message, content {}", content);
JSONObject jsonObject = JSONObject.parseObject(content);
......@@ -109,30 +78,31 @@ public class WeiXinEventConsumer {
if (event.equals("unsubscribe"))
this.restTemplateClient.unsubscribe(subscribeBean);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
} catch (Exception e) {
log.error("WXSubscribeConsumer || subOrUnSub msg error || {} || {}", content, e.getMessage());
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
if (MapUtils.isNotEmpty(error)) {
String errorStart = this.error.get("start");
if (errorStart.equalsIgnoreCase("true")) {
String fileName = this.error.get("fileName")+"_"+LocalDate.now() +".log";
String filePath = this.error.get("filePath");
String filePath1 = filePath+fileName;
FileUtil.writeStringToFile2(filePath1, content, e.getMessage());
}
}
/**
* @description 添加收藏记录
* @param content 消息内容
*/
/*@RabbitHandler
@RabbitListener(bindings = {
@QueueBinding(value = @Queue(value = RabbitMqConfig.COLLECTION_ADD_QUEUE),
exchange = @Exchange(value = ExchangeTypes.DIRECT))},
containerFactory = "managementRabbitListenerContainerFactory")*/
@Transactional
public void addCollection(String content) {
try {
log.info("receive UserCollection add message, content {}", content);
this.restTemplateClient.addCollection(content);
} catch (Exception e) {
log.error("CollectionAddConsumer || UserCollection add error || {}", e.toString(), e);
e.printStackTrace();
log.info("ucEventConsumer ====>>>> end");
}
}
}
......
package com.topdraw.mq.domain;
import lombok.Getter;
// 实体类型
@Getter
public enum EntityType {
// 用户
PERSON,
// 会员
MEMBER,
// 小屏-微信
WEIXIN_USER,
// 大屏
VIS_USER,
// 地址
PERSON_ADDRESS
}
package com.topdraw.mq.domain;
// 关注的事件
public enum EventType {
// 关注
SUBSCRIBE,
// 取消关注
UN_SUBSCRIBE,
// 绑定
BIND,
// 取消绑定
UN_BIND,
// 变更主账号
C_M_ACCOUNT,
// 获取积分
GAIN_POINT,
// 消耗积分
CONSUME_POINT,
// 观影
VIEWING
}
......@@ -173,4 +173,19 @@ public class RestTemplateClient {
log.info("uc response: " + entityBody);*/
return null;
}
public String dealViewRecord(String content) {
String url = BASE_URL + "/uce/userOperation/addCollection";
//处理接口调用 中文不显示问题
content = new String(Base64.getEncoder().encode(content.getBytes(StandardCharsets.UTF_8)));
restTemplate.postForEntity(url, content, String.class);
/* ResponseEntity<String> responseEntity = restTemplate.postForEntity(url, content, String.class);
String entityBody = "";
if (responseEntity.getStatusCode().is2xxSuccessful()) {
entityBody = responseEntity.getBody();
}
log.info("uc response: " + entityBody);*/
return null;
}
}
......
package com.topdraw.util;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.time.LocalDateTime;
/**
* @author ：
* @description：
* @function :
* @date ：Created in 2022/5/12 11:43
* @version: ：
* @modified By：
* @since : modified in 2022/5/12 11:43
*/
public class FileUtil {
private static void createFile(String filePath){
File testFile = new File(filePath);
File fileParent = testFile.getParentFile();//返回的是File类型,可以调用exsit()等方法
//String fileParentPath = testFile.getParent();//返回的是String类型
if (!fileParent.exists()) {
fileParent.mkdirs();// 能创建多级目录
}
if (!testFile.exists()) {
try {
testFile.createNewFile();//有路径才能创建文件
} catch (IOException e) {
e.printStackTrace();
}
}
}
public static void writeStringToFile2(String filePath, String content, String error) {
try {
String property = System.getProperty("user.dir");
filePath = property + filePath;
createFile(filePath);
FileWriter fw = new FileWriter(filePath, true);
BufferedWriter bw = new BufferedWriter(fw);
bw.append(LocalDateTime.now()+"\n");
bw.write("【content】==>> \n"+content+"\n");
bw.write("【error】==>> \n"+error+"\n");
bw.write("----------------------------------------------------------------\n");
bw.close();
fw.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}
package com.topdraw.util;
import java.util.ArrayList;
import java.util.List;
/**
* @author ：
* @description：
* @function :
* @date ：Created in 2022/5/5 12:57
* @version: ：
* @modified By：
* @since : modified in 2022/5/5 12:57
*/
public class ListUtil {
public static List<List<Integer>> sortList(List<List<Integer>> list, String desc) {
return null;
}
public static boolean containsValue(List<Integer> list, Integer value) {
return false;
}
public static void main(String[] args) {
List<List<Integer>> listList = new ArrayList<>();
List<Integer> integerList = new ArrayList<>();
integerList.add(1);
integerList.add(30);
List<Integer> integerList1 = new ArrayList<>();
integerList1.add(31);
integerList1.add(60);
listList.add(integerList);
listList.add(integerList1);
// Integer firstMaxInteger = listList.get(0).get;
/*Integer firstMaxInteger = 0;
for (List<Integer> integers : listList) {
firstMaxInteger = integers.get(integerList.size() - 1);
}*/
}
}
package com.topdraw.util;
import org.redisson.api.RLock;
public class RedissonUtil {
public static void lock(RLock rLock){
rLock.lock();
}
public static void unlock(RLock rLock){
if (rLock.isLocked() && rLock.isHeldByCurrentThread())
rLock.unlock();
}
}
spring:
# 数据源
datasource:
# 数据源地址
# url: jdbc:log4jdbc:mysql://139.196.192.242:3306/tj_user_0819?serverTimezone=Asia/Shanghai&characterEncoding=utf8&useSSL=false
# # 用户名
# username: root
# # 密码
# password: Tjlh@2017
url: jdbc:log4jdbc:mysql://122.112.214.149:3306/tj_user_iptv?serverTimezone=Asia/Shanghai&characterEncoding=utf8&useSSL=false
username: root
password: root
......@@ -70,47 +64,92 @@ mutil-mq:
service:
# ip
host: 122.112.214.149
# host: 139.196.145.150
# 端口
port: 5672
# 用户名
username: guest
# username: admin
# 密码
password: guest
# password: Topdraw1qaz
# 虚拟空间
virtual-host: member_center
# host: 139.196.145.150 # rabbitmq的连接地址
# port: 5672 # rabbitmq的连接端口号
# virtual-host: member_center # rabbitmq的虚拟hosthhh
# username: admin # rabbitmq的用户名
# password: Topdraw1qaz # rabbitmq的密码
virtual-host: member_center_chongshu
publisher-confirms: true #如果对异步消息需要回调必须设置为true
# 管理侧
management:
# ip
# host: 139.196.145.150
# # 端口
# port: 5672
# # 用户名
# username: admin
# # 密码
# password: Topdraw1qaz
# # 虚拟空间
# virtual-host: member_center
# host: 122.112.214.149 # rabbitmq的连接地址
host: 122.112.214.149 # rabbitmq的连接地址
port: 5672 # rabbitmq的连接端口号
virtual-host: member_center # rabbitmq的虚拟hosthhh
virtual-host: none # rabbitmq的虚拟host
username: guest # rabbitmq的用户名
password: guest # rabbitmq的密码
# username: admin # rabbitmq的用户名
# password: Topdraw1qaz # rabbitmq的密码
publisher-confirms: true #如果对异步消息需要回调必须设置为true
# 服务属性
service:
#平台类型 service: 服务侧 management: 管理侧
platform: service
# 服务域 mobile:小屏侧 vis:大屏侧
type: vis
mq:
list:
- source: event
exchange: event.exchange
queue: event.queue
exchange-type: direct
routing-key:
active:
- source: collection
exchange: collection.exchange
queue: collection.queue
exchange-type: direct
routing-key:
active:
- source: viewRecord
exchange: viewRecord.exchange
queue: viewRecord.queue
exchange-type: direct
routing-key:
active:
- source: eventBus
exchange: uc.eventbus
queue: uc.eventbus
exchange-type: topic
routing-key: uc.eventbus.*.topic
active:
- source: uce
exchange: uce.exchange
queue: uce.queue
exchange-type: direct
routing-key:
active: service
- source: wechat
exchange: wechat.exchange
queue: weixin.subOrUnSub.queue
exchange-type: direct
routing-key:
active:
error:
logs:
list:
- type: eventBus
filePath: /logs/mq/eventBus/
fileName: error
start: on
- type: ucg
filePath: /logs/mq/ucg/
fileName: error
start: on
- type: uce
filePath: /logs/mq/uce/
fileName: error
start: on
- type: wechat
filePath: /logs/mq/wechat/
fileName: error
start: on
# uc-engine服务地址
api:
baseUrl: http://127.0.0.1:8447
......
......@@ -2,8 +2,16 @@ package com.topdraw.resttemplate;
import com.alibaba.fastjson.JSONObject;
import com.topdraw.BaseTest;
import org.apache.poi.hssf.usermodel.HSSFWorkbook;
import org.apache.poi.util.IOUtils;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import java.io.*;
import java.lang.reflect.Field;
import java.time.LocalDate;
import java.time.LocalDateTime;
public class RestTemplateTest extends BaseTest {
......@@ -16,4 +24,92 @@ public class RestTemplateTest extends BaseTest {
System.out.println(memberInfo);
}
@Test
public void error(){
String msg3 = "{\n" +
" \"evt\": \"play\", \n" +
" \"deviceType\": 1, \n" +
// " \"time\": \"2022-04-14 00:10:09\",\n" +
" \"time\": \"2022-05-03 23:10:09\",\n" +
" \"msgData\": {\n" +
" \"platformAccount\": \"topdraw\", \n" +
" \"playDuration\": "+23+", \n" +
" \"mediaCode\": \"\", \n" +
" \"mediaName\": \"\"\n" +
" }\n" +
" }";
String classpathUrlPrefix = ApplicationContext.CLASSPATH_URL_PREFIX;
String panfu = "";
// String filePath = classpathUrlPrefix+"/mq/play/error.log";
String property = System.getProperty("user.dir");
String separator = File.separator;
System.out.println(property);
String fileName = "error_"+ LocalDate.now() +".log";
String filePath = property+separator+"logs"+separator+"mq"+separator+fileName;
createFile(filePath);
this.writeStringToFile2(filePath, msg3);
String a = "{\n" +
"\"data\": [\n" +
"\t{\n" +
"\t\"app_id\": 57,\n" +
"\t\"user_id\": 1,\n" +
"\t\"type\": 1,\n" +
"\t\"name\": \"PersonalCollectionRecords\",\n" +
"\t\"count\": 22,\n" +
"\t\"images\": \"{\\\"map\\\":{\\\"poster\\\":[0]},\\\"list\\\":[{\\\"id\\\":47422,\\\"type\\\":2,\\\"width\\\":222,\\\"height\\\":294,\\\"fileUrl\\\":\\\"upload/image/media/2020-07-30/9a8a02db-9444-4bff-ba54-ea784ae4f88c.jpg\\\",\\\"size\\\":104643}]}\",\n" +
"\t\"id\": 756756,\n" +
"\t\"user_collection_id\": 1,\n" +
"\t\"detail_folder_code\": \"Default\",\n" +
"\t\"detail_type\": \"MEDIA\",\n" +
"\t\"detail_id\": 46532,\n" +
"\t\"detail_code\": \"media_558bc45a-5480-46ec-be9a-c749ffdbdf49\",\n" +
"\t\"detail_name\": \"熊出没之探险日记2\",\n" +
"\t\"detail_total_index\": 40,\n" +
"\t\"detail_sequence\": 1,\n" +
"\t\"create_time\": 1644503167000,\n" +
"\t\"update_time\": 1644503167000\n" +
"\t}\n" +
"\t],\n" +
"\"platformAccount\": \"topdraw\"\n" +
"}";
}
private void createFile(String filePath){
File testFile = new File(filePath);
File fileParent = testFile.getParentFile();//返回的是File类型,可以调用exsit()等方法
String fileParentPath = testFile.getParent();//返回的是String类型
System.out.println("fileParent:" + fileParent);
System.out.println("fileParentPath:" + fileParentPath);
if (!fileParent.exists()) {
fileParent.mkdirs();// 能创建多级目录
}
if (!testFile.exists()) {
try {
testFile.createNewFile();//有路径才能创建文件
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println(testFile);
}
public void writeStringToFile2(String filePath, String error) {
try {
FileWriter fw = new FileWriter(filePath, true);
BufferedWriter bw = new BufferedWriter(fw);
bw.append(LocalDateTime.now()+"=>"+error+"\n");
//bw.write("我是");// 往已有的文件上添加字符串
//bw.write("程序猿\n ");
bw.close();
fw.close();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
......