Commit 0c06e9e7 authored by jiaorz's avatar jiaorz

Merge branch 'master-activity-bk' into dev

# Conflicts:
#	xx-activity/xx-activity-server/src/main/java/com/xxfc/platform/activity/biz/ActivityPopularizeBiz.java
#	xx-uccn/xx-uccn-server/src/main/java/com/xxfc/platform/uccn/rest/VehicleModelController.java
parents 6f22d9ea 6d621aee
......@@ -19,14 +19,12 @@ import com.xxfc.platform.universal.feign.MQSenderFeign;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.math.BigDecimal;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static com.github.wxiaoqi.security.common.constant.CommonConstants.SYS_FALSE;
import static com.github.wxiaoqi.security.common.constant.CommonConstants.SYS_TRUE;
......@@ -86,13 +84,13 @@ public class ActivityPopularizeBiz extends BaseBiz<ActivityPopularizeMapper, Act
}});
log.info("活动邀请注册:activityPopularize = {}", activityPopularize.toString());
if (activityPopularize != null) {
String lockKey = String.format("%s%d%d", registerQueueDTO.getInParamDTO().getActivityCode(), appUserDTO.getInviterAccount(), activityPopularize.getId());
RLock rLock = redissonLock.getRLock(lockKey);
try {
boolean isSuccess = rLock.tryLock(5, 5, TimeUnit.SECONDS);
if (isSuccess) {
// String lockKey = String.format("%s%d%d", registerQueueDTO.getInParamDTO().getActivityCode(), appUserDTO.getInviterAccount(), activityPopularize.getId());
// RLock rLock = redissonLock.getRLock(lockKey);
// try {
// boolean isSuccess = rLock.tryLock(5, 5, TimeUnit.SECONDS);
// if (isSuccess) {
log.info("1, {}", System.currentTimeMillis());
Thread.sleep(1000);
//Thread.sleep(1000);
log.info("2, {}", System.currentTimeMillis());
List<ActivityPopularizeItem> activityPopularizeItems = activityPopularizeItemBiz.selectByPopularizeId(activityPopularize.getId());
ActivityPopularizeItem activityPopularizeItem = activityPopularizeItems.get(activityPopularizeItems.size() - 1);
......@@ -206,16 +204,15 @@ public class ActivityPopularizeBiz extends BaseBiz<ActivityPopularizeMapper, Act
}}));
}
}
} else {
// 获取锁失败
log.info("tryLock fail, key = [{}]", lockKey);
}
} catch (InterruptedException e) {
log.error("tryLock fail, key = [{}]", lockKey);
} finally {
log.info("执行结束, {}", System.currentTimeMillis());
}
// } else {
// // 获取锁失败
// log.info("tryLock fail, key = [{}]", lockKey);
// }
// } catch (InterruptedException e) {
// log.error("tryLock fail, key = [{}]", lockKey);
// } finally {
// log.info("执行结束, {}", System.currentTimeMillis());
// }
}
}
}
......
......@@ -18,8 +18,8 @@ import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.*;
import java.util.concurrent.locks.StampedLock;
@Component
@Slf4j
......@@ -30,28 +30,39 @@ public class PopularizeMQHandler {
@Autowired
ActivityUserJoinBiz activityUserJoinBiz;
// 同时并发执行的线程数
public static int threadTotal = 10;
final Semaphore semaphore = new Semaphore(threadTotal);
final CountDownLatch countDownLatch = new CountDownLatch(10);
private final static StampedLock lock = new StampedLock();
//@RabbitListener(queues = {RabbitActivityConfig.POPULARZIE_REGISTER_QUEUE,RabbitActivityConfig.POPULARZIE_AUTH_QUEUE})
@RabbitListener(queues = {RabbitActivityConfig.POPULARZIE_0101_QUEUE})
public void popularizeHandler(Message message, @Headers Map<String, Object> headers, Channel channel) {
ExecutorService executorService = Executors.newCachedThreadPool();
executorService.execute(new Runnable() {
@Override
public void run() {
executorService.execute(() -> {
try {
semaphore.acquire();
String messageId = message.getMessageProperties().getMessageId();
String msg = new String(message.getBody(), "UTF-8");
log.info("接收到的消息:msg = {}, 消息ID是:messageId = {} ", msg, messageId);
if (StringUtils.isNotBlank(msg)) {
long stamp = lock.writeLock();
try {
RegisterQueueDTO registerQueueDTO = JSONUtil.toBean(msg, RegisterQueueDTO.class);
activityPopularizeBiz.handleRegister(registerQueueDTO);
} finally {
lock.unlock(stamp);
}
}
executorService.shutdown();
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
// 手动签收
channel.basicAck(deliveryTag, false);
semaphore.release();
} catch (Exception e) {
log.info("接收到的消息失败");
try {
......@@ -61,7 +72,8 @@ public class PopularizeMQHandler {
}
log.error(e.getMessage(), e);
}
}
executorService.shutdown();
countDownLatch.countDown();
});
}
//新人注册有礼
......
......@@ -55,6 +55,7 @@ public class VehicleModelController extends CommonBaseController {
return objectRestResponse;
}
@ApiOperation("车型详情")
@GetMapping(value = "/detail")
@IgnoreUserToken
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment