本文深入解析了 LangChain4j 框架下的并发处理机制与线程安全问题。内容涵盖多层级并发控制架构（应用层、HTTP 层、API 层），详细阐述了 HTTP 客户端配置、模型层并发控制及异步/响应式编程支持。重点分析了核心组件的线程安全性，包括无状态与有状态组件的区别，以及 ChatMemory 和 EmbeddingModel 的具体实现方案。此外，文章提供了连接池、线程池配置的最佳实践，性能监控与压力测试方法，并总结了面试中关于并发处理的核心要点与常见线程安全问题的解决方案。
LangChain4j 采用多层级并发控制策略：
┌───────────────────────────────────────────────────┐
│ 应用层并发控制 │
├───────────────────────────────────────────────────┤
│ 线程池配置 | 请求队列 | 超时控制 | 熔断机制 │
├───────────────────────────────────────────────────┤
│ HTTP 客户端并发控制 │
├───────────────────────────────────────────────────┤
│ 连接池管理 | 连接复用 | 请求复用 | 流控机制 │
├───────────────────────────────────────────────────┤
│ 供应商 API 并发限制 │
├───────────────────────────────────────────────────┤
│ 速率限制 | 令牌限制 | 配额管理 | 优先级队列 │
└───────────────────────────────────────────────────┘
// LangChain4j 并发相关的关键接口和类
public interface ConcurrentModel {
// 异步模型接口
interface AsyncChatModel extends ChatLanguageModel {
CompletableFuture<Response<String>> generateAsync(String prompt);
CompletableFuture<Response<String>> generateAsync(List<ChatMessage> messages);
}
// 流式模型接口
interface StreamingChatModel {
Flux<Response<String>> generateStream(List<ChatMessage> messages);
}
// 批量处理接口
interface BatchChatModel {
List<Response<String>> generateBatch(List<List<ChatMessage>> batches);
}
}
package com.example.concurrent;
import dev.langchain4j.model.openai.OpenAiChatModel;
import okhttp3.*;
import java.util.concurrent.*;
/**
* 高级 HTTP 客户端并发配置
*/
public class ConcurrentHttpClientConfig {
/**
* 配置高性能 HTTP 客户端
*/
public static OkHttpClient createHighPerformanceClient() {
// 1. 连接池配置
ConnectionPool connectionPool = new ConnectionPool(
100, // 最大空闲连接数
5, TimeUnit.MINUTES // 连接保持时间
);
// 2. 调度器配置（用于异步请求）
Dispatcher dispatcher = new Dispatcher();
dispatcher.setMaxRequests(200); // 最大并发请求数
dispatcher.setMaxRequestsPerHost(50); // 每主机最大并发数
// 3. 连接超时和读写超时
OkHttpClient.Builder builder = new OkHttpClient.Builder()
.connectTimeout(30, TimeUnit.SECONDS) // 连接超时
.writeTimeout(60, TimeUnit.SECONDS) // 写入超时
.readTimeout(120, TimeUnit.SECONDS) // 读取超时
.callTimeout(300, TimeUnit.SECONDS) // 总调用超时
.connectionPool(connectionPool)
.dispatcher(dispatcher);
// 4. 拦截器配置（用于监控和重试）
builder.addInterceptor(new RetryInterceptor(3)); // 重试拦截器
builder.addInterceptor(new RateLimitInterceptor()); // 限流拦截器
builder.addInterceptor(new LoggingInterceptor()); // 日志拦截器
// 5. 连接复用配置
builder.connectionSpecs(Arrays.asList(
ConnectionSpec.MODERN_TLS, // 现代 TLS 连接
ConnectionSpec.COMPATIBLE_TLS // 兼容 TLS 连接
));
// 6. DNS 优化
builder.dns(new DnsOverHttps.Builder()
.url(HttpUrl.get("https://cloudflare-dns.com/dns-query"))
.build());
return builder.build();
}
/**
* 重试拦截器
*/
static class RetryInterceptor implements Interceptor {
private final int maxRetries;
public RetryInterceptor(int maxRetries) {
this.maxRetries = maxRetries;
}
@Override
public Response intercept(Chain chain) throws IOException {
Request request = chain.request();
Response response = null;
IOException exception = null;
for (int retryCount = 0; retryCount <= maxRetries; retryCount++) {
try {
response = chain.proceed(request);
// 检查是否应该重试
if (shouldRetry(response, retryCount)) {
response.close();
if (retryCount < maxRetries) {
waitBeforeRetry(retryCount);
continue;
}
}
return response;
} catch (IOException e) {
exception = e;
if (retryCount < maxRetries) {
waitBeforeRetry(retryCount);
continue;
}
}
}
throw exception != null ? exception : new IOException("Request failed");
}
private boolean shouldRetry(Response response, int retryCount) {
int code = response.code();
return code == 429 || // 太多请求
code >= 500 || // 服务器错误
code == 408; // 请求超时
}
private void waitBeforeRetry(int retryCount) {
try {
long waitTime = (long) Math.pow(2, retryCount) * 1000;
Thread.sleep(waitTime);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
/**
* 限流拦截器
*/
static class RateLimitInterceptor implements Interceptor {
private final RateLimiter rateLimiter;
public RateLimitInterceptor() {
// 每秒 10 个请求，突发 20 个
this.rateLimiter = RateLimiter.create(10.0, 20);
}
@Override
public Response intercept(Chain chain) throws IOException {
rateLimiter.acquire(); // 获取令牌
return chain.proceed(chain.request());
}
}
}
package com.example.concurrent;
import dev.langchain4j.model.chat.ChatLanguageModel;
import dev.langchain4j.model.output.Response;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
/**
* 并发安全的聊天模型包装器
*/
public class ConcurrentChatModel implements ChatLanguageModel {
private final ChatLanguageModel delegate;
private final ExecutorService executorService;
private final Semaphore concurrencyLimiter;
private final AtomicLong requestCounter;
private final ConcurrentHashMap<String, AtomicLong> requestMetrics;
private final CircuitBreaker circuitBreaker;
public ConcurrentChatModel(ChatLanguageModel delegate, int maxConcurrency, int queueSize) {
this.delegate = delegate;
// 1. 配置线程池
this.executorService = new ThreadPoolExecutor(
maxConcurrency, // 核心线程数
maxConcurrency * 2, // 最大线程数
60L, TimeUnit.SECONDS, // 空闲线程超时
new LinkedBlockingQueue<>(queueSize), // 工作队列
new ThreadFactory() {
private ();
Thread {
(r, + counter.incrementAndGet());
t.setDaemon();
t;
}
},
.CallerRunsPolicy()
);
.concurrencyLimiter = (maxConcurrency);
.requestCounter = ();
.requestMetrics = <>();
.circuitBreaker = (, , );
}
Response<String> {
generate(Collections.singletonList(UserMessage.from(prompt)));
}
Response<String> {
(!circuitBreaker.allowRequest()) {
();
}
System.nanoTime();
UUID.randomUUID().toString();
{
(!concurrencyLimiter.tryAcquire(, TimeUnit.SECONDS)) {
();
}
CompletableFuture<Response<String>> future = CompletableFuture.supplyAsync(() -> {
{
recordRequestStart(requestId);
delegate.generate(messages);
} (Exception e) {
circuitBreaker.recordFailure();
e;
}
}, executorService);
Response<String> response = future.get(, TimeUnit.SECONDS);
circuitBreaker.recordSuccess();
recordRequestSuccess(requestId, startTime);
response;
} (TimeoutException e) {
circuitBreaker.recordFailure();
(, e);
} (ExecutionException e) {
circuitBreaker.recordFailure();
(, e.getCause());
} (Exception e) {
circuitBreaker.recordFailure();
(, e);
} {
concurrencyLimiter.release();
}
}
List<Response<String>> {
IntStream.range(, n).parallel()
.mapToObj(i -> generate(messages))
.collect(Collectors.toList());
}
Stream<Response<String>> {
StreamSupport.stream(
Spliterators.spliteratorUnknownSize( <Response<String>>() {
;
{
hasNext;
}
Response<String> {
;
}
}, Spliterator.ORDERED), );
}
CompletableFuture<Response<String>> {
CompletableFuture.supplyAsync(() -> generate(messages), executorService);
}
CompletableFuture<List<Response<String>>> {
List<CompletableFuture<Response<String>>> futures = batches.stream()
.map(::generateAsync)
.collect(Collectors.toList());
CompletableFuture.allOf(futures.toArray( []))
.thenApply(v -> futures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList()));
}
{
requestCounter.incrementAndGet();
requestMetrics.put(requestId, (System.nanoTime()));
}
{
System.nanoTime() - startTime;
requestMetrics.get(requestId);
(counter != ) {
counter.set(duration);
}
}
Map<String, Object> {
Map<String, Object> metrics = <>();
metrics.put(, requestCounter.get());
metrics.put(, concurrencyLimiter.availablePermits());
metrics.put(, ((ThreadPoolExecutor) executorService).getQueue().size());
metrics.put(, ((ThreadPoolExecutor) executorService).getActiveCount());
List<Long> durations = requestMetrics.values().stream()
.map(AtomicLong::get)
.filter(v -> v > )
.collect(Collectors.toList());
(!durations.isEmpty()) {
durations.stream()
.mapToLong(Long::longValue)
.average()
.orElse() / ;
metrics.put(, avgDuration);
}
metrics;
}
{
failureThreshold;
recoveryTimeout;
failureRateThreshold;
();
();
;
;
{
.failureThreshold = failureThreshold;
.recoveryTimeout = recoveryTimeout;
.failureRateThreshold = failureRateThreshold;
}
{
(!isOpen) {
;
}
(System.currentTimeMillis() - lastFailureTime > recoveryTimeout) {
isOpen = ;
failureCount.set();
successCount.set();
;
}
;
}
{
successCount.incrementAndGet();
calculateFailureRate();
(failureRate < failureRateThreshold) {
failureCount.set();
}
}
{
failureCount.incrementAndGet();
lastFailureTime = System.currentTimeMillis();
(failures >= failureThreshold) {
isOpen = ;
}
}
{
successCount.get() + failureCount.get();
total > ? () failureCount.get() / total : ;
}
}
}
package com.example.concurrent;
import dev.langchain4j.model.chat.ChatLanguageModel;
import dev.langchain4j.model.output.Response;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.util.List;
import java.util.concurrent.CompletableFuture;
/**
* 响应式聊天模型包装器
*/
public class ReactiveChatModel {
private final ChatLanguageModel delegate;
private final Scheduler scheduler;
public ReactiveChatModel(ChatLanguageModel delegate) {
this.delegate = delegate;
// 配置响应式调度器
this.scheduler = Schedulers.newBoundedElastic(100, 1000, "ReactiveChatModel", 60, true);
}
/**
* 响应式生成方法
*/
public Mono<Response<String>> generateReactive(List<ChatMessage> messages) {
return Mono.fromCallable(() -> delegate.generate(messages))
.subscribeOn(scheduler)
.timeout(Duration.ofSeconds(120))
.onErrorResume(e -> {
// 错误处理逻辑
return Mono.error(new RuntimeException("Generation failed", e));
});
}
/**
* 流式响应式生成
*/
public Flux<String> generateStreamReactive(List<ChatMessage> messages) {
Flux.create(fluxSink -> {
{
String[] chunks = {, , };
(String chunk : chunks) {
(fluxSink.isCancelled()) {
;
}
fluxSink.next(chunk);
{
Thread.sleep();
} (InterruptedException e) {
fluxSink.error(e);
;
}
}
fluxSink.complete();
} (Exception e) {
fluxSink.error(e);
}
}).subscribeOn(scheduler);
}
Flux<Response<String>> {
Flux.fromIterable(batches)
.parallel()
.runOn(Schedulers.parallel())
.flatMap(::generateReactive)
.sequential();
}
Flux<Response<String>> {
messageFlux
.onBackpressureBuffer()
.flatMap(::generateReactive, );
}
}
package com.example.threadsafe;
import java.util.concurrent.atomic.*;
import java.util.concurrent.locks.*;
/**
* LangChain4j 组件线程安全性分析
*/
public class ThreadSafetyAnalysis {
/**
* 1. 无状态组件的线程安全性
*/
public static class StatelessComponent {
// 无状态组件是天然线程安全的
public String process(String input) {
return input.toUpperCase();
}
}
/**
* 2. 有状态组件的线程安全性挑战
*/
public static class StatefulComponent {
// 问题：非线程安全的计数器
private int counter = 0;
public int increment() {
counter++; // 非原子操作，存在竞态条件
return counter;
}
// 解决方案 1：使用原子类
private AtomicInteger safeCounter = new AtomicInteger(0);
public int incrementSafely() {
return safeCounter.incrementAndGet();
}
}
/**
* 3. 缓存组件的线程安全性
*/
public <K, V> {
ConcurrentHashMap<K, V> cache = <>();
();
V {
lock.readLock().lock();
{
cache.get(key);
} {
lock.readLock().unlock();
}
}
{
lock.writeLock().lock();
{
cache.put(key, value);
} {
lock.writeLock().unlock();
}
}
V {
cache.computeIfAbsent(key, mappingFunction);
}
}
{
ThreadLocal<Session> threadLocalSession = <>();
ThreadLocal<Session> managedThreadLocalSession = <Session>() {
Session {
();
}
{
.get().close();
.remove();
}
};
ConcurrentHashMap<String, Session> sessionMap = <>();
Session {
sessionMap.computeIfAbsent(sessionId, id -> (id));
}
}
{
List<Tool> tools;
();
Object {
toolLock.lock();
{
findTool(toolName);
tool.execute(input);
} {
toolLock.unlock();
}
}
ConcurrentHashMap<String, ReentrantLock> toolLocks = <>();
Object {
toolLocks.computeIfAbsent(toolName, k -> ());
lock.lock();
{
findTool(toolName);
tool.execute(input);
} {
lock.unlock();
}
}
}
}
package com.example.threadsafe;
import dev.langchain4j.memory.ChatMemory;
import dev.langchain4j.memory.chat.MessageWindowChatMemory;
/**
* ChatMemory 线程安全分析
*/
public class ChatMemoryThreadSafety {
/**
* InMemoryChatMemory 的线程安全问题
*/
public static void analyzeInMemoryChatMemory() {
// 1. MessageWindowChatMemory 不是线程安全的
ChatMemory memory = MessageWindowChatMemory.withMaxMessages(10);
// 在多线程环境中可能出现的问题：
// - ConcurrentModificationException
// - 消息丢失
// - 状态不一致
// 解决方案：使用同步包装器
ChatMemory synchronizedMemory = synchronizeChatMemory(memory);
}
/**
* 创建线程安全的 ChatMemory 包装器
*/
public static ChatMemory synchronizeChatMemory(ChatMemory delegate) {
return new ChatMemory() {
private final Object lock = new Object();
@Override
public String id() {
synchronized (lock) {
return delegate.id();
}
}
{
(lock) {
delegate.add(message);
}
}
List<ChatMessage> {
(lock) {
<>(delegate.messages());
}
}
{
(lock) {
delegate.clear();
}
}
};
}
{
String sessionId;
RedisTemplate<String, Object> redisTemplate;
();
{
.sessionId = sessionId;
.redisTemplate = redisTemplate;
}
{
+ sessionId;
redisTemplate.execute( <Object>() {
Object DataAccessException {
operations.watch(key);
List<ChatMessage> messages = (List<ChatMessage>) operations.opsForValue().get(key);
(messages == ) {
messages = <>();
}
messages.add(message);
operations.multi();
operations.opsForValue().set(key, messages);
operations.exec();
}
});
}
}
}
package com.example.threadsafe;
import dev.langchain4j.model.embedding.EmbeddingModel;
import dev.langchain4j.model.output.Response;
import java.util.List;
import java.util.concurrent.*;
/**
* EmbeddingModel 并发处理
*/
public class EmbeddingModelConcurrency {
/**
* 线程安全的 EmbeddingModel 包装器
*/
public static class ThreadSafeEmbeddingModel implements EmbeddingModel {
private final EmbeddingModel delegate;
private final ExecutorService executorService;
private final RateLimiter rateLimiter;
public ThreadSafeEmbeddingModel(EmbeddingModel delegate, int maxConcurrency) {
this.delegate = delegate;
// 配置线程池
this.executorService = new ThreadPoolExecutor(
maxConcurrency,
maxConcurrency * 2,
60L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(1000),
new ThreadPoolExecutor.CallerRunsPolicy()
);
// 配置限流器（根据 API 限制）
this.rateLimiter = RateLimiter.create(100.0); // 每秒 100 个请求
}
@Override
public Response<Embedding> {
embedAll(Collections.singletonList(text)).content().get();
}
Response<List<Embedding>> {
rateLimiter.acquire(texts.size());
List<CompletableFuture<Response<Embedding>>> futures = texts.stream()
.map(text -> CompletableFuture.supplyAsync(() -> delegate.embed(text), executorService))
.collect(Collectors.toList());
List<Embedding> embeddings = futures.stream()
.map(CompletableFuture::join)
.map(Response::content)
.collect(Collectors.toList());
Response.from(embeddings);
}
Response<List<Embedding>> {
;
List<List<String>> batches = <>();
( ; i < texts.size(); i += batchSize) {
batches.add(texts.subList(i, Math.min(i + batchSize, texts.size())));
}
List<Embedding> allEmbeddings = batches.parallelStream()
.flatMap(batch -> {
rateLimiter.acquire(batch.size());
delegate.embedAll(batch).content().stream();
})
.collect(Collectors.toList());
Response.from(allEmbeddings);
}
}
}
package com.example.bestpractice;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
/**
* HTTP 连接池最佳实践
*/
public class ConnectionPoolBestPractice {
public static CloseableHttpClient createOptimalHttpClient() {
// 1. 连接管理器配置
PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
// 设置总连接数
connectionManager.setMaxTotal(200);
// 设置每个路由的最大连接数
connectionManager.setDefaultMaxPerRoute(50);
// 设置连接存活时间
connectionManager.setValidateAfterInactivity(30000); // 30 秒
// 2. 请求配置
RequestConfig requestConfig = RequestConfig.custom()
.setConnectTimeout(30000) // 连接超时
.setSocketTimeout(60000) // Socket 超时
.setConnectionRequestTimeout(5000) // 从连接池获取连接超时
.build();
// 3. 构建 HTTP 客户端
return HttpClientBuilder.create()
.setConnectionManager(connectionManager)
.setDefaultRequestConfig(requestConfig)
.setRetryHandler(new DefaultHttpRequestRetryHandler(3, ))
.disableCookieManagement()
.setKeepAliveStrategy( ())
.build();
}
{
PoolingHttpClientConnectionManager connectionManager;
ScheduledExecutorService monitorExecutor;
{
.connectionManager = connectionManager;
.monitorExecutor = Executors.newSingleThreadScheduledExecutor();
startMonitoring();
}
{
monitorExecutor.scheduleAtFixedRate(() -> {
{
connectionManager.getTotalStats();
System.out.println();
System.out.println( + stats.getAvailable());
System.out.println( + stats.getLeased());
System.out.println( + stats.getPending());
System.out.println( + stats.getMax());
(stats.getAvailable() < ) {
System.err.println();
}
connectionManager.closeExpiredConnections();
connectionManager.closeIdleConnections(, TimeUnit.SECONDS);
} (Exception e) {
System.err.println( + e.getMessage());
}
}, , , TimeUnit.SECONDS);
}
}
}
package com.example.bestpractice;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 异步编程最佳实践
*/
public class AsyncBestPractice {
/**
* 1. 配置合适的线程池
*/
public static ExecutorService createOptimalThreadPool() {
int corePoolSize = Runtime.getRuntime().availableProcessors();
int maxPoolSize = corePoolSize * 2;
return new ThreadPoolExecutor(
corePoolSize,
maxPoolSize,
60L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(10000),
new ThreadFactory() {
private final AtomicInteger counter = new AtomicInteger();
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r, "LLM-Worker-" + counter.incrementAndGet());
t.setDaemon(true);
t.setUncaughtExceptionHandler((thread, throwable) -> {
System.err.println( + thread.getName() + + throwable.getMessage());
});
t;
}
},
.AbortPolicy()
);
}
{
CompletableFuture<String> {
CompletableFuture<String> timeoutFuture = <>();
CompletableFuture.runAsync(() -> {
{
Thread.sleep(unit.toMillis(timeout));
timeoutFuture.completeExceptionally( ());
} (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
future.applyToEither(timeoutFuture, Function.identity());
}
CompletableFuture<List<String>> {
(maxConcurrency);
List<CompletableFuture<String>> controlledFutures = futures.stream()
.map(future -> CompletableFuture.supplyAsync(() -> {
{
semaphore.acquire();
future.get();
} (InterruptedException e) {
Thread.currentThread().interrupt();
(e);
} (ExecutionException e) {
(e);
} {
semaphore.release();
}
}))
.collect(Collectors.toList());
CompletableFuture.allOf(controlledFutures.toArray( []))
.thenApply(v -> controlledFutures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList()));
}
}
{
Flux<String> {
source
.onBackpressureBuffer(, BufferOverflowStrategy.DROP_LATEST)
.publishOn(Schedulers.parallel())
.doOnNext(item -> {
})
.doOnError(error -> {
})
.doOnComplete(() -> {
});
}
Mono<String> {
mono
.retryWhen(Retry.backoff(, Duration.ofSeconds())
.maxBackoff(Duration.ofSeconds())
.jitter()
.doAfterRetry(retrySignal -> {
}))
.timeout(Duration.ofSeconds())
.onErrorResume(error -> {
Mono.just();
});
}
}
}
package com.example.monitoring;
import io.micrometer.core.instrument.*;
import io.micrometer.core.instrument.binder.jvm.*;
import io.micrometer.core.instrument.binder.system.*;
import io.micrometer.prometheus.PrometheusConfig;
import io.micrometer.prometheus.PrometheusMeterRegistry;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
/**
* LangChain4j 性能监控
*/
public class PerformanceMonitor {
private final MeterRegistry meterRegistry;
private final ConcurrentHashMap<String, Timer> timers;
private final AtomicLong totalRequests;
private final AtomicLong failedRequests;
private final AtomicLong activeRequests;
public PerformanceMonitor() {
this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
this.timers = new ConcurrentHashMap<>();
this.totalRequests = new AtomicLong();
this.failedRequests = new AtomicLong();
this.activeRequests = new AtomicLong();
// 注册指标
registerMetrics();
}
private void registerMetrics() {
().bindTo(meterRegistry);
().bindTo(meterRegistry);
().bindTo(meterRegistry);
().bindTo(meterRegistry);
().bindTo(meterRegistry);
().bindTo(meterRegistry);
Gauge.builder(, totalRequests, AtomicLong::get)
.description()
.register(meterRegistry);
Gauge.builder(, activeRequests, AtomicLong::get)
.description()
.register(meterRegistry);
Gauge.builder(, failedRequests, AtomicLong::get)
.description()
.register(meterRegistry);
}
Timer.Sample {
activeRequests.incrementAndGet();
Timer.start(meterRegistry);
}
{
totalRequests.incrementAndGet();
activeRequests.decrementAndGet();
timers.computeIfAbsent(operation, op ->
Timer.builder()
.tag(, op)
.tag(, )
.publishPercentiles(, , )
.register(meterRegistry)
);
sample.stop(timer);
}
{
totalRequests.incrementAndGet();
failedRequests.incrementAndGet();
activeRequests.decrementAndGet();
timers.computeIfAbsent(operation + , op ->
Timer.builder()
.tag(, operation)
.tag(, )
.tag(, error)
.register(meterRegistry)
);
sample.stop(timer);
}
{
Counter.builder()
.tag(, )
.register(meterRegistry)
.increment(inputTokens);
Counter.builder()
.tag(, )
.register(meterRegistry)
.increment(outputTokens);
}
String {
((PrometheusMeterRegistry) meterRegistry).scrape();
}
}
package com.example.testing;
import dev.langchain4j.model.chat.ChatLanguageModel;
import org.openjdk.jmh.annotations.*;
import org.openjdk.jmh.runner.Runner;
import org.openjdk.jmh.runner.RunnerException;
import org.openjdk.jmh.runner.options.Options;
import org.openjdk.jmh.runner.options.OptionsBuilder;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
/**
* LangChain4j 并发性能测试
*/
@State(Scope.Benchmark)
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@Warmup(iterations = 3, time = 5)
@Measurement(iterations = 5, time = 10)
@Fork(2)
@Threads(10) // 并发线程数
public class ConcurrentPerformanceTest {
private ChatLanguageModel model;
private ExecutorService executorService;
private CountDownLatch latch;
private AtomicInteger successCount;
private AtomicInteger failureCount;
@Setup
public void setup() {
// 初始化模型
model = createChatModel();
// 初始化线程池
executorService = Executors.newFixedThreadPool(50);
successCount = new AtomicInteger();
failureCount = new AtomicInteger();
}
@Benchmark
@BenchmarkMode({Mode.Throughput, Mode.AverageTime})
public void InterruptedException {
;
latch = (concurrentRequests);
( ; i < concurrentRequests; i++) {
executorService.submit(() -> {
{
model.generate( + Thread.currentThread().getId());
successCount.incrementAndGet();
} (Exception e) {
failureCount.incrementAndGet();
} {
latch.countDown();
}
});
}
latch.await(, TimeUnit.SECONDS);
}
{
executorService.shutdown();
System.out.println( + successCount.get());
System.out.println( + failureCount.get());
System.out.println( + () successCount.get() / (successCount.get() + failureCount.get()));
}
RunnerException {
()
.include(ConcurrentPerformanceTest.class.getSimpleName())
.build();
(options).run();
}
}
内存可见性问题：
// 问题：volatile 变量不保证原子性
private volatile int counter = 0;
public void increment() {
counter++; // 不是原子操作
}
// 解决方案：使用原子类
private AtomicInteger counter = new AtomicInteger(0);
public void increment() {
counter.incrementAndGet();
}
死锁风险：
// 避免嵌套锁
synchronized(lockA) {
synchronized(lockB) {
// 可能导致死锁
// ...
}
}
// 解决方案：锁排序
Object firstLock = lockA.hashCode() < lockB.hashCode() ? lockA : lockB;
Object secondLock = firstLock == lockA ? lockB : lockA;
synchronized(firstLock) {
synchronized(secondLock) {
// ...
}
}
竞态条件：
// 问题：非原子操作
if (!initialized) {
initialize(); // 可能被多个线程同时执行
initialized = true;
}
// 解决方案：双重检查锁
if (!initialized) {
synchronized(this) {
if (!initialized) {
initialize();
initialized = true;
}
}
}
Q: LangChain4j 如何处理并发请求？
A: LangChain4j 通过多层级并发控制机制：
Q: LangChain4j 是否存在线程安全问题？
A: 需要具体情况分析：
Q: 如何优化 LangChain4j 的并发性能？
A: 可以从以下几个方面优化：
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
查找任何按下的键的javascript键代码、代码、位置和修饰符。 在线工具，Keycode 信息在线工具，online
JavaScript 字符串转义/反转义；Java 风格 \uXXXX（Native2Ascii）编码与解码。 在线工具，Escape 与 Native 编解码在线工具，online
使用 Prettier 在浏览器内格式化 JavaScript 或 HTML 片段。 在线工具，JavaScript / HTML 格式化在线工具，online
Terser 压缩、变量名混淆，或 javascript-obfuscator 高强度混淆（体积会增大）。 在线工具，JavaScript 压缩与混淆在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online