java八股-Redis Stream和RocketMQ实现的解决方案

Redis Stream方案：

ShortLinkStatsSaveProducer.java

package com.nageoffer.shortlink.project.mq.producer;import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;import java.util.Map;import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_TOPIC_KEY;/*** 短链接监控状态保存消息队列生产者* 公众号：马丁玩编程，回复：加群，添加马哥微信（备注：link）获取项目资料*/
@Component
@RequiredArgsConstructor
public class ShortLinkStatsSaveProducer {private final StringRedisTemplate stringRedisTemplate;/*** 发送延迟消费短链接统计*/public void send(Map<String, String> producerMap) {stringRedisTemplate.opsForStream().add(SHORT_LINK_STATS_STREAM_TOPIC_KEY, producerMap);}
}

ShortLinkStatsSaveConsumer.java

package com.nageoffer.shortlink.project.mq.consumer;import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.nageoffer.shortlink.project.common.convention.exception.ServiceException;
import com.nageoffer.shortlink.project.dao.entity.LinkAccessLogsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkAccessStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkBrowserStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkDeviceStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkLocaleStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkNetworkStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkOsStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkStatsTodayDO;
import com.nageoffer.shortlink.project.dao.entity.ShortLinkGotoDO;
import com.nageoffer.shortlink.project.dao.mapper.LinkAccessLogsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkAccessStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkBrowserStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkDeviceStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkLocaleStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkNetworkStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkOsStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkStatsTodayMapper;
import com.nageoffer.shortlink.project.dao.mapper.ShortLinkGotoMapper;
import com.nageoffer.shortlink.project.dao.mapper.ShortLinkMapper;
import com.nageoffer.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.nageoffer.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.LOCK_GID_UPDATE_KEY;
import static com.nageoffer.shortlink.project.common.constant.ShortLinkConstant.AMAP_REMOTE_URL;@Slf4j
@Component
@RequiredArgsConstructor
public class ShortLinkStatsSaveConsumer implements StreamListener<String, MapRecord<String, String, String>> {private final ShortLinkMapper shortLinkMapper;private final ShortLinkGotoMapper shortLinkGotoMapper;private final RedissonClient redissonClient;private final LinkAccessStatsMapper linkAccessStatsMapper;private final LinkLocaleStatsMapper linkLocaleStatsMapper;private final LinkOsStatsMapper linkOsStatsMapper;private final LinkBrowserStatsMapper linkBrowserStatsMapper;private final LinkAccessLogsMapper linkAccessLogsMapper;private final LinkDeviceStatsMapper linkDeviceStatsMapper;private final LinkNetworkStatsMapper linkNetworkStatsMapper;private final LinkStatsTodayMapper linkStatsTodayMapper;private final StringRedisTemplate stringRedisTemplate;private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;@Value("${short-link.stats.locale.amap-key}")private String statsLocaleAmapKey;@Overridepublic void onMessage(MapRecord<String, String, String> message) {String stream = message.getStream();RecordId id = message.getId();if (messageQueueIdempotentHandler.isMessageBeingConsumed(id.toString())) {// 判断当前的这个消息流程是否执行完成if (messageQueueIdempotentHandler.isAccomplish(id.toString())) {return;}throw new ServiceException("消息未完成流程，需要消息队列重试");}try {Map<String, String> producerMap = message.getValue();ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);actualSaveShortLinkStats(statsRecord);stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());} catch (Throwable ex) {// 某某某情况宕机了messageQueueIdempotentHandler.delMessageProcessed(id.toString());log.error("记录短链接监控消费异常", ex);throw ex;}messageQueueIdempotentHandler.setAccomplish(id.toString());}public void actualSaveShortLinkStats(ShortLinkStatsRecordDTO statsRecord) {String fullShortUrl = statsRecord.getFullShortUrl();RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));RLock rLock = readWriteLock.readLock();rLock.lock();try {LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class).eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);String gid = shortLinkGotoDO.getGid();Date currentDate = statsRecord.getCurrentDate();int hour = DateUtil.hour(currentDate, true);Week week = DateUtil.dayOfWeekEnum(currentDate);int weekValue = week.getIso8601Value();LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder().pv(1).uv(statsRecord.getUvFirstFlag() ? 1 : 0).uip(statsRecord.getUipFirstFlag() ? 1 : 0).hour(hour).weekday(weekValue).fullShortUrl(fullShortUrl).date(currentDate).build();linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);Map<String, Object> localeParamMap = new HashMap<>();localeParamMap.put("key", statsLocaleAmapKey);localeParamMap.put("ip", statsRecord.getRemoteAddr());String localeResultStr = HttpUtil.get(AMAP_REMOTE_URL, localeParamMap);JSONObject localeResultObj = JSON.parseObject(localeResultStr);String infoCode = localeResultObj.getString("infocode");String actualProvince = "未知";String actualCity = "未知";if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")) {String province = localeResultObj.getString("province");boolean unknownFlag = StrUtil.equals(province, "[]");LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder().province(actualProvince = unknownFlag ? actualProvince : province).city(actualCity = unknownFlag ? actualCity : localeResultObj.getString("city")).adcode(unknownFlag ? "未知" : localeResultObj.getString("adcode")).cnt(1).fullShortUrl(fullShortUrl).country("中国").date(currentDate).build();linkLocaleStatsMapper.shortLinkLocaleState(linkLocaleStatsDO);}LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder().os(statsRecord.getOs()).cnt(1).fullShortUrl(fullShortUrl).date(currentDate).build();linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder().browser(statsRecord.getBrowser()).cnt(1).fullShortUrl(fullShortUrl).date(currentDate).build();linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder().device(statsRecord.getDevice()).cnt(1).fullShortUrl(fullShortUrl).date(currentDate).build();linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder().network(statsRecord.getNetwork()).cnt(1).fullShortUrl(fullShortUrl).date(currentDate).build();linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder().user(statsRecord.getUv()).ip(statsRecord.getRemoteAddr()).browser(statsRecord.getBrowser()).os(statsRecord.getOs()).network(statsRecord.getNetwork()).device(statsRecord.getDevice()).locale(StrUtil.join("-", "中国", actualProvince, actualCity)).fullShortUrl(fullShortUrl).build();linkAccessLogsMapper.insert(linkAccessLogsDO);shortLinkMapper.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1 : 0, statsRecord.getUipFirstFlag() ? 1 : 0);LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder().todayPv(1).todayUv(statsRecord.getUvFirstFlag() ? 1 : 0).todayUip(statsRecord.getUipFirstFlag() ? 1 : 0).fullShortUrl(fullShortUrl).date(currentDate).build();linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);} finally {rLock.unlock();}}
}

RocketMQ方案

<rocketmq-spring-boot-starter.version>2.2.3</rocketmq-spring-boot-starter.version><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>${rocketmq-spring-boot-starter.version}</version>
</dependency>

yaml配置

rocketmq:name-server: 127.0.0.1:9876producer:group: short-link_project-service_stats-save_pgtopic: short-link_project-service_topicsend-message-timeout: 2000retry-times-when-send-failed: 1retry-times-when-send-async-failed: 1consumer:group: short-link_project-service_stats-save_cg

ShortLinkStatsSaveProducer.java

package com.nageoffer.shortlink.project.mq.producer;import com.alibaba.fastjson2.JSON;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;import java.util.Map;
import java.util.UUID;/*** 短链接监控状态保存消息队列生产者* 公众号：马丁玩编程，回复：加群，添加马哥微信（备注：link）获取项目资料*/
@Slf4j
@Component
@RequiredArgsConstructor
public class ShortLinkStatsSaveProducer {private final RocketMQTemplate rocketMQTemplate;@Value("${rocketmq.producer.topic}")private String statsSaveTopic;/*** 发送延迟消费短链接统计*/public void send(Map<String, String> producerMap) {String keys = UUID.randomUUID().toString();producerMap.put("keys", keys);Message<Map<String, String>> build = MessageBuilder.withPayload(producerMap).setHeader(MessageConst.PROPERTY_KEYS, keys).build();SendResult sendResult;try {sendResult = rocketMQTemplate.syncSend(statsSaveTopic, build, 2000L);log.info("[消息访问统计监控] 消息发送结果：{}，消息ID：{}，消息Keys：{}", sendResult.getSendStatus(), sendResult.getMsgId(), keys);} catch (Throwable ex) {log.error("[消息访问统计监控] 消息发送失败，消息体：{}", JSON.toJSONString(producerMap), ex);// 自定义行为...}}
}

ShortLinkStatsSaveConsumer.java

package com.nageoffer.shortlink.project.mq.consumer;import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.nageoffer.shortlink.project.common.convention.exception.ServiceException;
import com.nageoffer.shortlink.project.dao.entity.LinkAccessLogsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkAccessStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkBrowserStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkDeviceStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkLocaleStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkNetworkStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkOsStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkStatsTodayDO;
import com.nageoffer.shortlink.project.dao.entity.ShortLinkGotoDO;
import com.nageoffer.shortlink.project.dao.mapper.LinkAccessLogsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkAccessStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkBrowserStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkDeviceStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkLocaleStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkNetworkStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkOsStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkStatsTodayMapper;
import com.nageoffer.shortlink.project.dao.mapper.ShortLinkGotoMapper;
import com.nageoffer.shortlink.project.dao.mapper.ShortLinkMapper;
import com.nageoffer.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.nageoffer.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.LOCK_GID_UPDATE_KEY;
import static com.nageoffer.shortlink.project.common.constant.ShortLinkConstant.AMAP_REMOTE_URL;@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(topic = "${rocketmq.producer.topic}",consumerGroup = "${rocketmq.consumer.group}"
)
public class ShortLinkStatsSaveConsumer implements RocketMQListener<Map<String, String>> {private final ShortLinkMapper shortLinkMapper;private final ShortLinkGotoMapper shortLinkGotoMapper;private final RedissonClient redissonClient;private final LinkAccessStatsMapper linkAccessStatsMapper;private final LinkLocaleStatsMapper linkLocaleStatsMapper;private final LinkOsStatsMapper linkOsStatsMapper;private final LinkBrowserStatsMapper linkBrowserStatsMapper;private final LinkAccessLogsMapper linkAccessLogsMapper;private final LinkDeviceStatsMapper linkDeviceStatsMapper;private final LinkNetworkStatsMapper linkNetworkStatsMapper;private final LinkStatsTodayMapper linkStatsTodayMapper;private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;@Value("${short-link.stats.locale.amap-key}")private String statsLocaleAmapKey;@Overridepublic void onMessage(Map<String, String> producerMap) {String keys = producerMap.get("keys");if (!messageQueueIdempotentHandler.isMessageProcessed(keys)) {// 判断当前的这个消息流程是否执行完成if (messageQueueIdempotentHandler.isAccomplish(keys)) {return;}throw new ServiceException("消息未完成流程，需要消息队列重试");}try {String fullShortUrl = producerMap.get("fullShortUrl");if (StrUtil.isNotBlank(fullShortUrl)) {String gid = producerMap.get("gid");ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);actualSaveShortLinkStats(fullShortUrl, gid, statsRecord);}} catch (Throwable ex) {// 删除幂等标识messageQueueIdempotentHandler.delMessageProcessed(id.toString());log.error("记录短链接监控消费异常", ex);throw ex;}messageQueueIdempotentHandler.setAccomplish(keys);}public void actualSaveShortLinkStats(String fullShortUrl, String gid, ShortLinkStatsRecordDTO statsRecord) {fullShortUrl = Optional.ofNullable(fullShortUrl).orElse(statsRecord.getFullShortUrl());RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));RLock rLock = readWriteLock.readLock();rLock.lock();try {if (StrUtil.isBlank(gid)) {LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class).eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);gid = shortLinkGotoDO.getGid();}int hour = DateUtil.hour(new Date(), true);Week week = DateUtil.dayOfWeekEnum(new Date());int weekValue = week.getIso8601Value();LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder().pv(1).uv(statsRecord.getUvFirstFlag() ? 1 : 0).uip(statsRecord.getUipFirstFlag() ? 1 : 0).hour(hour).weekday(weekValue).fullShortUrl(fullShortUrl).gid(gid).date(new Date()).build();linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);Map<String, Object> localeParamMap = new HashMap<>();localeParamMap.put("key", statsLocaleAmapKey);localeParamMap.put("ip", statsRecord.getRemoteAddr());String localeResultStr = HttpUtil.get(AMAP_REMOTE_URL, localeParamMap);JSONObject localeResultObj = JSON.parseObject(localeResultStr);String infoCode = localeResultObj.getString("infocode");String actualProvince = "未知";String actualCity = "未知";if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")) {String province = localeResultObj.getString("province");boolean unknownFlag = StrUtil.equals(province, "[]");LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder().province(actualProvince = unknownFlag ? actualProvince : province).city(actualCity = unknownFlag ? actualCity : localeResultObj.getString("city")).adcode(unknownFlag ? "未知" : localeResultObj.getString("adcode")).cnt(1).fullShortUrl(fullShortUrl).country("中国").gid(gid).date(new Date()).build();linkLocaleStatsMapper.shortLinkLocaleState(linkLocaleStatsDO);}LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder().os(statsRecord.getOs()).cnt(1).gid(gid).fullShortUrl(fullShortUrl).date(new Date()).build();linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder().browser(statsRecord.getBrowser()).cnt(1).gid(gid).fullShortUrl(fullShortUrl).date(new Date()).build();linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder().device(statsRecord.getDevice()).cnt(1).gid(gid).fullShortUrl(fullShortUrl).date(new Date()).build();linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder().network(statsRecord.getNetwork()).cnt(1).gid(gid).fullShortUrl(fullShortUrl).date(new Date()).build();linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder().user(statsRecord.getUv()).ip(statsRecord.getRemoteAddr()).browser(statsRecord.getBrowser()).os(statsRecord.getOs()).network(statsRecord.getNetwork()).device(statsRecord.getDevice()).locale(StrUtil.join("-", "中国", actualProvince, actualCity)).gid(gid).fullShortUrl(fullShortUrl).build();linkAccessLogsMapper.insert(linkAccessLogsDO);shortLinkMapper.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1 : 0, statsRecord.getUipFirstFlag() ? 1 : 0);LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder().todayPv(1).todayUv(statsRecord.getUvFirstFlag() ? 1 : 0).todayUip(statsRecord.getUipFirstFlag() ? 1 : 0).gid(gid).fullShortUrl(fullShortUrl).date(new Date()).build();linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);} catch (Throwable ex) {log.error("短链接访问量统计异常", ex);} finally {rLock.unlock();}}
}