@Configuration
public class RabbitMQConfig {
    
    // 交换机和队列常量定义
    public static final String DEMO_EXCHANGE = "demo.direct.exchange";
    public static final String DEMO_QUEUE = "demo.message.queue";
    public static final String DEMO_ROUTING_KEY = "demo.routing.key";
    
    // 创建直连交换机
    @Bean
    public DirectExchange demoExchange() {
        return ExchangeBuilder
                .directExchange(DEMO_EXCHANGE)
                .durable(true)      // 持久化
                .build();
    }
    
    // 创建消息队列
    @Bean
    public Queue demoQueue() {
        return QueueBuilder
                .durable(DEMO_QUEUE)
                .withArgument("x-dead-letter-exchange", "")  // 死信交换机
                .withArgument("x-dead-letter-routing-key", "demo.dlq")
                .build();
    }
    
    // 绑定队列到交换机
    @Bean
    public Binding demoBinding() {
        return BindingBuilder
                .bind(demoQueue())
                .to(demoExchange())
                .with(DEMO_ROUTING_KEY);
    }
    
    // JSON消息转换器
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

@Service
@Slf4j
public class MessageProducerService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Autowired
    private RabbitMQConfig rabbitMQConfig;
    
    /**
     * 发送简单文本消息
     */
    public boolean sendTextMessage(String content) {
        try {
            rabbitTemplate.convertAndSend(
                rabbitMQConfig.DEMO_EXCHANGE,
                rabbitMQConfig.DEMO_ROUTING_KEY,
                content
            );
            log.info("消息发送成功: {}", content);
            return true;
        } catch (Exception e) {
            log.error("消息发送失败: {}", content, e);
            return false;
        }
    }
    
    /**
     * 发送对象消息（自动JSON序列化）
     */
    public void sendObjectMessage(MessageDTO messageDTO) {
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(UUID.randomUUID().toString());
        
        rabbitTemplate.convertAndSend(
            rabbitMQConfig.DEMO_EXCHANGE,
            rabbitMQConfig.DEMO_ROUTING_KEY,
            messageDTO,
            message -> {
                // 设置消息属性
                message.getMessageProperties().setMessageId(correlationData.getId());
                message.getMessageProperties().setTimestamp(new Date());
                message.getMessageProperties().setContentType("application/json");
                return message;
            },
            correlationData
        );
        
        log.info("对象消息发送完成，消息ID: {}", correlationData.getId());
    }
    
    /**
     * 带有确认回调的消息发送
     */
    @PostConstruct
    public void initConfirmCallback() {
        // 消息发送到Exchange确认回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息到达Exchange成功，消息ID: {}", correlationData.getId());
            } else {
                log.error("消息到达Exchange失败，消息ID: {}，原因: {}", 
                         correlationData.getId(), cause);
            }
        });
        
        // 消息路由到Queue失败回调
        rabbitTemplate.setReturnsCallback(returned -> {
            log.error("消息路由到队列失败: {}", returned.getMessage());
        });
    }
}

@Component
@Slf4j
public class MessageConsumerService {
    
    /**
     * 监听队列消息 - 自动确认模式
     */
    @RabbitListener(
        queues = RabbitMQConfig.DEMO_QUEUE,
        concurrency = "3-10"  // 动态调整消费者数量
    )
    public void handleMessage(String message) {
        log.info("消费者[{}]收到消息: {}", 
                Thread.currentThread().getName(), message);
        
        try {
            // 模拟业务处理
            processBusinessLogic(message);
            log.info("消息处理完成: {}", message);
        } catch (Exception e) {
            log.error("处理消息失败: {}", message, e);
            // 异常时根据策略决定是否重新入队
            throw new AmqpRejectAndDontRequeueException("处理失败，不重新入队");
        }
    }
    
    /**
     * 监听对象消息
     */
    @RabbitListener(queues = RabbitMQConfig.DEMO_QUEUE)
    public void handleObjectMessage(MessageDTO messageDTO) {
        log.info("收到对象消息: ID={}, Content={}", 
                messageDTO.getId(), messageDTO.getContent());
        
        // 处理对象消息
        // ...
    }
    
    /**
     * 手动ACK模式 - 更精确的控制
     */
    @RabbitListener(queues = RabbitMQConfig.DEMO_QUEUE)
    public void handleMessageWithAck(
            Message message, 
            Channel channel) throws IOException {
        
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        
        try {
            String msgBody = new String(message.getBody());
            log.info("处理消息（手动ACK）: {}", msgBody);
            
            // 业务处理
            boolean success = processBusinessLogic(msgBody);
            
            if (success) {
                // 确认消息
                channel.basicAck(deliveryTag, false);
                log.info("消息确认成功: {}", deliveryTag);
            } else {
                // 拒绝消息，不重新入队
                channel.basicNack(deliveryTag, false, false);
                log.warn("消息处理失败，已拒绝: {}", deliveryTag);
            }
            
        } catch (Exception e) {
            log.error("处理消息异常，消息ID: {}", deliveryTag, e);
            // 异常时拒绝消息
            channel.basicNack(deliveryTag, false, false);
        }
    }
    
    private void processBusinessLogic(String message) {
        // 实际业务逻辑处理
        // 这里可以调用其他Service进行业务处理
        log.debug("处理业务逻辑: {}", message);
        
        // 模拟处理耗时
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

@Component
@Slf4j
public class DeadLetterConsumer {
    
    @RabbitListener(queues = DeadLetterConfig.DLQ_QUEUE)
    public void handleDeadLetter(Message failedMessage) {
        log.error("收到死信消息: {}", new String(failedMessage.getBody()));
        log.error("原始路由Key: {}", failedMessage.getMessageProperties().getReceivedRoutingKey());
        log.error("失败原因: {}", failedMessage.getMessageProperties().getHeader("x-death"));
        
        // 这里可以实现死信消息的处理逻辑
        // 1. 记录到数据库
        // 2. 发送告警通知
        // 3. 人工干预处理
    }
}

@Data
public class MessageRequest {
    private String content;
    private String type;
}

@Data
public class ResponseResult<T> {
    private Integer code;
    private String message;
    private T data;
    
    public static <T> ResponseResult<T> success(T data) {
        ResponseResult<T> result = new ResponseResult<>();
        result.setCode(200);
        result.setMessage("success");
        result.setData(data);
        return result;
    }
    
    public static ResponseResult<?> success(String message) {
        ResponseResult<?> result = new ResponseResult<>();
        result.setCode(200);
        result.setMessage(message);
        return result;
    }
}

@Transactional(rollbackFor = Exception.class)
public void update(Long modelId) throws GseException {
    synchronized (this) {
        //业务逻辑代码
    }
}

@Transactional(rollbackFor = Exception.class)
public void update(Long modelId) throws GseException {
    RLock lock = redissonClient.getLock(REDIS_COST_MODEL_ID_LOCK + modelId);
    try {
        if (lock.tryLock(20, 2, TimeUnit.SECONDS)) {
            //业务逻辑代码
        } else {
            log.error("获取分布式锁失败");
        }
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    } finally {
        // 判断当前线程是否持有锁
        if (lock.isHeldByCurrentThread()) {
            //释放当前锁
            lock.unlock();
        }
    }
}

   public void addLock(Long modelId) throws GseException {
    synchronized (this) {
        update(modelId);
    }
}

@Transactional(rollbackFor = Exception.class)
public void xxxCopy(Long modelId) throws GseException {
    synchronized (this) {
        //业务逻辑代码
    }
}

public void addLock(Long modelId) throws GseException {
  RLock lock = redissonClient.getLock(REDIS_COST_MODEL_ID_LOCK + modelId);
  try {
      if (lock.tryLock(20, 2, TimeUnit.SECONDS)) {
          update(modelId);
      } else {
          log.error("获取分布式锁失败");
      }
  } catch (InterruptedException e) {
      throw new RuntimeException(e);
  } finally {
      // 判断当前线程是否持有锁
      if (lock.isHeldByCurrentThread()) {
          //释放当前锁
          lock.unlock();
      }
  }
}

@Transactional(rollbackFor = Exception.class)
public void update(Long modelId) throws GseException {
    synchronized (this) {
        //业务逻辑代码
    }
}

@Transactional
public void updateWithLock(Long id) {
    // 使用SELECT FOR UPDATE锁定行
    MyData data = myDataMapper.selectForUpdate(id);

    // 检查状态
    if (data.getStatus() == TARGET_STATUS) {
        // 执行更新
        data.setStatus(NEW_STATUS);
        myDataMapper.update(data);
    }
}

public boolean updateWithVersion(Long id) {
    MyData data = myDataMapper.findById(id);

    if (data.getStatus() != TARGET_STATUS) {
        return false;
    }

    // 带版本号的更新
    int rows = myDataMapper.updateWithVersion(
            id,
            NEW_STATUS,
            data.getVersion(),  // 当前版本
            data.getVersion() + 1 // 新版本
    );

    return rows > 0;
}

<update id="updateWithVersion">
UPDATE my_table
SET status = #{newStatus},
version = #{newVersion}
WHERE id = #{id}
AND version = #{oldVersion}
</update>

import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.concurrent.TimeUnit;

@Service
public class VisitStatisticsService {

    private final RedisTemplate<String, String> redisTemplate;

    public VisitStatisticsService(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    // 获取当日PV的key
    private String getTodayPVKey() {
        return "stat:pv:" + LocalDate.now().toString();
    }

    // 获取当日UV的key
    private String getTodayUVKey() {
        return "stat:uv:" + LocalDate.now().toString();
    }

    // 获取当天剩余的秒数（到24点的秒数）
    private long getRemainingSecondsToday() {
        ZonedDateTime now = ZonedDateTime.now(ZoneId.systemDefault());
        ZonedDateTime midnight = now.toLocalDate().plusDays(1).atStartOfDay(ZoneId.systemDefault());
        return midnight.toEpochSecond() - now.toEpochSecond();
    }

    /**
     * 记录访问
     * @param userId 用户ID（可以为IP地址或其他唯一标识）
     */
    public void recordVisit(String userId) {
        // 获取当天剩余的秒数
        long remainingSeconds = getRemainingSecondsToday();
        
        // 记录PV（访问量）
        ValueOperations<String, String> valueOps = redisTemplate.opsForValue();
        String pvKey = getTodayPVKey();
        valueOps.increment(pvKey);
        redisTemplate.expire(pvKey, remainingSeconds, TimeUnit.SECONDS);
        
        // 记录UV（独立访客）
        String uvKey = getTodayUVKey();
        redisTemplate.opsForHyperLogLog().add(uvKey, userId);
        redisTemplate.expire(uvKey, remainingSeconds, TimeUnit.SECONDS);
    }

    /**
     * 获取当日访问量(PV)
     */
    public Long getTodayPV() {
        String pvKey = getTodayPVKey();
        String value = redisTemplate.opsForValue().get(pvKey);
        return value == null ? 0L : Long.parseLong(value);
    }

    /**
     * 获取当日访问人数(UV)
     */
    public Long getTodayUV() {
        String uvKey = getTodayUVKey();
        return redisTemplate.opsForHyperLogLog().size(uvKey);
    }

    /**
     * 获取历史某天的访问量(PV)
     * @param date 日期
     */
    public Long getPVByDate(LocalDate date) {
        String pvKey = "stat:pv:" + date.toString();
        String value = redisTemplate.opsForValue().get(pvKey);
        return value == null ? 0L : Long.parseLong(value);
    }

    /**
     * 获取历史某天的访问人数(UV)
     * @param date 日期
     */
    public Long getUVByDate(LocalDate date) {
        String uvKey = "stat:uv:" + date.toString();
        return redisTemplate.opsForHyperLogLog().size(uvKey);
    }
}

@RestController
@RequestMapping("/api/visit")
public class VisitController {

    private final VisitStatisticsService visitStatisticsService;

    public VisitController(VisitStatisticsService visitStatisticsService) {
        this.visitStatisticsService = visitStatisticsService;
    }

    @GetMapping("/record")
    public String recordVisit(@RequestParam String userId) {
        visitStatisticsService.recordVisit(userId);
        return "Visit recorded for user: " + userId;
    }

    @GetMapping("/stats")
    public Map<String, Object> getStats() {
        Map<String, Object> result = new HashMap<>();
        result.put("pv", visitStatisticsService.getTodayPV());
        result.put("uv", visitStatisticsService.getTodayUV());
        return result;
    }
}