20250701

-XX:+UseG1GC
-XX:MaxGCPauseMillis=200  # 目标最大停顿时间
-XX:InitiatingHeapOccupancyPercent=45  # 老年代占用45%时触发GC
-XX:G1HeapRegionSize=8m  # 调整Region大小

-XX:+UseZGC
-XX:MaxHeapSize=4g  # 根据实际需求设置
-XX:ConcGCThreads=4  # 并发GC线程数
-XX:ZCollectionInterval=1000  # GC周期（毫秒）五

// 示例代码
public void synchronizedMethod() {
    synchronized (this) {
        // 同步代码块
    }
}

// 对应的字节码片段
monitorenter        // 进入同步块
// 业务逻辑
monitorexit         // 正常退出同步块
monitorexit         // 异常退出同步块（通过Exception table触发）

// 非公平锁获取锁逻辑
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {  // CAS操作获取锁
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {  // 可重入
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

public class DynamicThreadPool {
    private final ThreadPoolExecutor executor;
    
    public DynamicThreadPool() {
        executor = new ThreadPoolExecutor(
            10,                     // 初始核心线程数
            100,                    // 初始最大线程数
            60, TimeUnit.SECONDS,   // 空闲线程存活时间
            new LinkedBlockingQueue<>(1000)  // 任务队列
        );
    }
    
    // 动态调整核心线程数
    public void setCorePoolSize(int corePoolSize) {
        executor.setCorePoolSize(corePoolSize);
    }
    
    // 动态调整最大线程数
    public void setMaximumPoolSize(int maximumPoolSize) {
        executor.setMaximumPoolSize(maximumPoolSize);
    }
    
    // 动态调整队列容量（需自定义队列）
    public void setQueueCapacity(int capacity) {
        if (executor.getQueue() instanceof ResizableBlockingQueue) {
            ((ResizableBlockingQueue) executor.getQueue()).setCapacity(capacity);
        }
    }
}

public class ResizableBlockingQueue<E> extends LinkedBlockingQueue<E> {
    private final AtomicInteger capacity;
    
    public ResizableBlockingQueue(int capacity) {
        super(capacity);
        this.capacity = new AtomicInteger(capacity);
    }
    
    public void setCapacity(int newCapacity) {
        int oldCapacity = capacity.getAndSet(newCapacity);
        // 可添加队列大小调整的额外逻辑
    }
    
    @Override
    public int remainingCapacity() {
        return capacity.get() - size();
    }
}

public class AutoScalingThreadPool {
    private final ScheduledExecutorService monitorService;
    private final ThreadPoolExecutor executor;
    
    public AutoScalingThreadPool() {
        executor = new ThreadPoolExecutor(10, 100, 60, TimeUnit.SECONDS, 
                new LinkedBlockingQueue<>(1000));
        
        monitorService = Executors.newSingleThreadScheduledExecutor();
        // 每5秒监控一次
        monitorService.scheduleAtFixedRate(this::monitorAndAdjust, 0, 5, TimeUnit.SECONDS);
    }
    
    private void monitorAndAdjust() {
        // 获取当前线程池状态
        int activeCount = executor.getActiveCount();
        int queueSize = executor.getQueue().size();
        long completedTasks = executor.getCompletedTaskCount();
        
        // 根据负载调整参数（示例策略）
        if (queueSize > 800) {  // 队列接近满时
            executor.setCorePoolSize(Math.min(executor.getCorePoolSize() + 5, 100));
        } else if (activeCount < 5 && executor.getCorePoolSize() > 10) {  // 负载低时
            executor.setCorePoolSize(Math.max(executor.getCorePoolSize() - 5, 10));
        }
    }
    
    public void shutdown() {
        monitorService.shutdown();
        executor.shutdown();
    }
}

public class ConfigCenterThreadPool {
    private final ThreadPoolExecutor executor;
    private final ConfigService configService;  // 配置中心客户端
    
    public ConfigCenterThreadPool() {
        // 初始化线程池，使用配置中心的初始值
        int coreSize = getConfig("threadPool.coreSize", 10);
        int maxSize = getConfig("threadPool.maxSize", 100);
        int queueSize = getConfig("threadPool.queueSize", 1000);
        
        executor = new ThreadPoolExecutor(coreSize, maxSize, 60, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(queueSize));
        
        // 注册配置变更监听器
        configService.addListener("threadPool.properties", this::onConfigChanged);
    }
    
    private void onConfigChanged(ConfigChangeEvent event) {
        if (event.isChanged("threadPool.coreSize")) {
            int newCoreSize = event.getNewValue("threadPool.coreSize");
            executor.setCorePoolSize(newCoreSize);
        }
        
        if (event.isChanged("threadPool.maxSize")) {
            int newMaxSize = event.getNewValue("threadPool.maxSize");
            executor.setMaximumPoolSize(newMaxSize);
        }
        
        if (event.isChanged("threadPool.queueSize") && 
            executor.getQueue() instanceof ResizableBlockingQueue) {
            int newQueueSize = event.getNewValue("threadPool.queueSize");
            ((ResizableBlockingQueue<?>) executor.getQueue()).setCapacity(newQueueSize);
        }
    }
    
    private int getConfig(String key, int defaultValue) {
        // 从配置中心获取值，带默认值
        return configService.getIntProperty(key, defaultValue);
    }
}

// 线程池预热示例
public void warmup() {
    for (int i = 0; i < executor.getCorePoolSize(); i++) {
        executor.execute(() -> {
            // 执行空任务或轻量级任务
        });
    }
}

// 服务元数据模型
public class ServiceInstance {
    private String serviceId;       // 服务唯一标识
    private String instanceId;      // 实例唯一标识
    private String host;            // 主机IP
    private int port;               // 端口
    private boolean secure;         // 是否安全通信
    private Map<String, String> metadata;  // 自定义元数据
    private long registrationTime;  // 注册时间
    private HealthStatus status;    // 健康状态
    // getters/setters
}

// 健康状态枚举
public enum HealthStatus {
    UP, DOWN, STARTING, OUT_OF_SERVICE, UNKNOWN
}

// 服务注册接口
public interface ServiceRegistry {
    void register(ServiceInstance instance);  // 注册服务
    void deregister(String instanceId);       // 注销服务
    void heartbeat(String instanceId);        // 发送心跳
}

// 服务发现接口
public interface ServiceDiscovery {
    List<ServiceInstance> getInstances(String serviceId);  // 获取服务实例列表
    List<String> getServices();                            // 获取所有服务ID
}

// 基于ZooKeeper的实现示例
public class ZookeeperServiceRegistry implements ServiceRegistry, ServiceDiscovery {
    private final CuratorFramework client;
    private static final String BASE_PATH = "/services";
    
    public ZookeeperServiceRegistry(String zkConnectString) {
        this.client = CuratorFrameworkFactory.newClient(
            zkConnectString, 
            15000,  // 会话超时
            5000,   // 连接超时
            new ExponentialBackoffRetry(1000, 3)  // 重试策略
        );
        client.start();
    }
    
    @Override
    public void register(ServiceInstance instance) {
        String path = getServicePath(instance.getServiceId()) + "/" + instance.getInstanceId();
        try {
            client.create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.EPHEMERAL)  // 临时节点，会话断开自动删除
                .forPath(path, toJsonBytes(instance));
        } catch (Exception e) {
            throw new RuntimeException("Failed to register service", e);
        }
    }
    
    // 其他方法实现...
}

// 健康检查器接口
public interface HealthChecker {
    HealthStatus check(ServiceInstance instance);  // 检查服务健康状态
}

// 基于HTTP的健康检查实现
public class HttpHealthChecker implements HealthChecker {
    private final RestTemplate restTemplate;
    private static final int TIMEOUT = 3000;  // 超时时间3秒
    
    public HttpHealthChecker() {
        this.restTemplate = new RestTemplate();
        // 配置超时
        ClientHttpRequestFactory factory = restTemplate.getRequestFactory();
        if (factory instanceof SimpleClientHttpRequestFactory) {
            ((SimpleClientHttpRequestFactory) factory).setConnectTimeout(TIMEOUT);
            ((SimpleClientHttpRequestFactory) factory).setReadTimeout(TIMEOUT);
        }
    }
    
    @Override
    public HealthStatus check(ServiceInstance instance) {
        try {
            String healthUrl = buildHealthUrl(instance);
            ResponseEntity<String> response = restTemplate.getForEntity(healthUrl, String.class);
            return response.getStatusCode().is2xxSuccessful() ? HealthStatus.UP : HealthStatus.DOWN;
        } catch (Exception e) {
            return HealthStatus.DOWN;
        }
    }
    
    private String buildHealthUrl(ServiceInstance instance) {
        String scheme = instance.isSecure() ? "https" : "http";
        return scheme + "://" + instance.getHost() + ":" + instance.getPort() + "/actuator/health";
    }
}

// 服务注册中心主类
public class ServiceRegistryCenter {
    private final ServiceRegistry registry;
    private final ServiceDiscovery discovery;
    private final HealthChecker healthChecker;
    private final EventExecutorGroup bossGroup;
    private final EventExecutorGroup workerGroup;
    private Channel serverChannel;
    
    public ServiceRegistryCenter(String zkConnectString, int port) {
        this.registry = new ZookeeperServiceRegistry(zkConnectString);
        this.discovery = new ZookeeperServiceDiscovery(zkConnectString);
        this.healthChecker = new HttpHealthChecker();
        
        // 初始化Netty服务器
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(
                        new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 4),
                        new LengthFieldPrepender(4),
                        new JsonDecoder(),
                        new JsonEncoder(),
                        new RegistryServerHandler(registry, discovery)
                    );
                }
            });
        
        // 启动服务器
        try {
            ChannelFuture future = bootstrap.bind(port).sync();
            serverChannel = future.channel();
            System.out.println("Service Registry Center started on port " + port);
            
            // 启动健康检查定时任务
            ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(1);
            scheduledExecutor.scheduleAtFixedRate(this::checkServicesHealth, 30, 30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    
    private void checkServicesHealth() {
        List<String> serviceIds = discovery.getServices();
        for (String serviceId : serviceIds) {
            List<ServiceInstance> instances = discovery.getInstances(serviceId);
            for (ServiceInstance instance : instances) {
                HealthStatus status = healthChecker.check(instance);
                if (status != instance.getStatus()) {
                    instance.setStatus(status);
                    // 更新服务状态
                    registry.updateStatus(instance.getInstanceId(), status);
                    System.out.println("Service " + instance.getInstanceId() + " status changed to " + status);
                }
            }
        }
    }
    
    public void shutdown() {
        if (serverChannel != null) {
            serverChannel.close();
        }
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
    
    public static void main(String[] args) {
        new ServiceRegistryCenter("localhost:2181", 8888);
    }
}

// 同步发送示例（确保消息发送成功）
SendResult result = producer.send(msg, 3000);  // 超时时间3秒
if (result.getSendStatus() == SendStatus.SEND_OK) {
    // 消息发送成功
}

producer.setRetryTimesWhenSendFailed(3);  // 失败重试3次

TransactionMQProducer producer = new TransactionMQProducer("group");
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务
        return LocalTransactionState.COMMIT_MESSAGE;
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 事务状态回查
        return LocalTransactionState.COMMIT_MESSAGE;
    }
});

# broker.conf配置
flushDiskType = SYNC_FLUSH  # 同步刷盘，确保消息写入磁盘后才返回成功

# broker.conf配置
brokerRole = SYNC_MASTER  # 同步双写模式
flushDiskType = SYNC_FLUSH

// 消费者配置
consumer.setMessageModel(MessageModel.CLUSTERING);  // 集群模式
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        try {
            // 处理消息
            processMessage(msgs);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  // 手动提交
        } catch (Exception e) {
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;  // 失败重试
        }
    }
});

// 幂等消费示例（基于业务ID去重）
public void processMessage(MessageExt msg) {
    String bizId = msg.getUserProperty("bizId");
    if (redisClient.sismember("processed_msgs", bizId)) {
        return;  // 已处理，直接返回
    }
    
    // 处理业务逻辑
    processBiz(msg);
    
    // 标记为已处理
    redisClient.sadd("processed_msgs", bizId);
}

// 推荐配置
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("nameserver:9876");
producer.setRetryTimesWhenSendFailed(3);  // 同步发送失败重试3次
producer.setRetryTimesWhenSendAsyncFailed(3);  // 异步发送失败重试3次
producer.setSendMsgTimeout(5000);  // 发送超时5秒

# broker.conf推荐配置
brokerRole = SYNC_MASTER  # 同步双写模式
flushDiskType = SYNC_FLUSH  # 同步刷盘
deleteWhen = 04  # 凌晨4点删除过期文件
fileReservedTime = 48  # 保留48小时
messageStorePlugIn = com.xxx.MyStorePlugin  # 可选：自定义存储插件增强可靠性

// 推荐配置
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
consumer.setNamesrvAddr("nameserver:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);  // 从最后位置消费
consumer.setConsumeThreadMin(20);  // 最小消费线程数
consumer.setConsumeThreadMax(50);  // 最大消费线程数
consumer.setMaxReconsumeTimes(16);  // 最大重试次数

if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end

public class RedlockExample {
    private static final int N = 5;  // 节点数
    private final List<Jedis> jedisClients;
    private static final String LOCK_KEY = "distributed_lock";
    private static final int LOCK_EXPIRE_MS = 10000;  // 锁有效期10秒
    
    public RedlockExample() {
        // 初始化5个Redis节点
        jedisClients = Arrays.asList(
            new Jedis("redis1", 6379),
            new Jedis("redis2", 6379),
            new Jedis("redis3", 6379),
            new Jedis("redis4", 6379),
            new Jedis("redis5", 6379)
        );
    }
    
    public boolean acquireLock(String requestId) {
        int acquiredNodes = 0;
        long startTime = System.currentTimeMillis();
        
        // 尝试锁定所有节点
        for (Jedis client : jedisClients) {
            try {
                // 使用SETNX + PX原子操作
                String result = client.set(LOCK_KEY, requestId, "NX", "PX", LOCK_EXPIRE_MS);
                if ("OK".equals(result)) {
                    acquiredNodes++;
                }
            } catch (Exception e) {
                // 忽略异常，继续尝试其他节点
            }
        }
        
        // 计算获取锁耗时
        long elapsedTime = System.currentTimeMillis() - startTime;
        // 检查是否成功锁定多数节点，且耗时小于锁有效期
        return acquiredNodes >= (N / 2 + 1) && elapsedTime < LOCK_EXPIRE_MS;
    }
    
    public void releaseLock(String requestId) {
        // 使用Lua脚本释放锁，确保原子性
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                        "return redis.call('del', KEYS[1]) else return 0 end";
        
        for (Jedis client : jedisClients) {
            try {
                client.eval(script, Collections.singletonList(LOCK_KEY), 
                           Collections.singletonList(requestId));
            } catch (Exception e) {
                // 忽略异常，继续释放其他节点
            }
        }
    }
}

非叶子节点: [10, 20, 30]
          /   |   |   \
叶子节点: [1, 5, 8] [12, 15, 18] [22, 25, 28] [32, 35, 38]

 Hash Table:
[0] -> (key1, value1) -> (key4, value4)
[1] -> (key2, value2)
[2] -> (key3, value3)

CREATE INDEX idx_age ON users(age);  -- B+树索引