{"id":74626,"date":"2026-02-10T09:25:27","date_gmt":"2026-02-10T01:25:27","guid":{"rendered":"https:\/\/www.wsisp.com\/helps\/74626.html"},"modified":"2026-02-10T09:25:27","modified_gmt":"2026-02-10T01:25:27","slug":"%e9%ab%98%e7%ba%a7java%e6%af%8f%e6%97%a5%e4%b8%80%e9%81%93%e9%9d%a2%e8%af%95%e9%a2%98-2025%e5%b9%b46%e6%9c%8826%e6%97%a5-%e5%9f%ba%e7%a1%80%e7%af%87langchain4j-%e4%bb%80%e4%b9%88%e6%98%af-tokenstrea","status":"publish","type":"post","link":"https:\/\/www.wsisp.com\/helps\/74626.html","title":{"rendered":"\u9ad8\u7ea7java\u6bcf\u65e5\u4e00\u9053\u9762\u8bd5\u9898-2025\u5e746\u670826\u65e5-\u57fa\u7840\u7bc7[LangChain4j]-\u4ec0\u4e48\u662f TokenStream\uff1f\u5982\u4f55\u6b63\u786e\u5904\u7406\u6d41\u5f0f\u54cd\u5e94\u4e2d\u7684\u5f02\u5e38\u548c\u4e2d\u65ad\uff1f"},"content":{"rendered":"

TokenStream \u4e0e\u6d41\u5f0f\u54cd\u5e94\u5f02\u5e38\u5904\u7406\u8be6\u89e3<\/h2>\n

\u4e00\u3001TokenStream \u6df1\u5ea6\u89e3\u6790<\/h3>\n

1.1 TokenStream \u7684\u6982\u5ff5\u4e0e\u8bbe\u8ba1<\/h4>\n

\/**
\n * TokenStream \u6838\u5fc3\u6982\u5ff5\u89e3\u6790
\n *
\n * \u5728 LLM \u6d41\u5f0f\u54cd\u5e94\u4e2d，TokenStream \u4ee3\u8868\u4e00\u4e2a**\u5b57\u7b26\u5e8f\u5217\u7684\u5f02\u6b65\u6d41**
\n * \u7c7b\u4f3c Java \u7684 Stream API，\u4f46\u4e13\u4e3a AI \u54cd\u5e94\u8bbe\u8ba1
\n *\/<\/span>
\npublic<\/span> class<\/span> TokenStreamConcept<\/span> {<\/span><\/p>\n

\/\/ TokenStream \u7684\u6838\u5fc3\u7279\u6027<\/span>
\n public<\/span> interface<\/span> TokenStreamCharacteristics<\/span> {<\/span>
\n \/**
\n * 1. \u5f02\u6b65\u6027：\u975e\u963b\u585e\u7684\u6570\u636e\u6d41
\n * 2. \u589e\u91cf\u6027：\u9010\u6b65\u751f\u6210\u548c\u6d88\u8d39
\n * 3. \u8fde\u7eed\u6027：\u4fdd\u6301\u8bed\u4e49\u8fde\u8d2f
\n * 4. \u53ef\u4e2d\u65ad\u6027：\u5141\u8bb8\u4e2d\u9014\u53d6\u6d88
\n * 5. \u9519\u8bef\u4f20\u64ad：\u652f\u6301\u5f02\u5e38\u4f20\u9012
\n *\/<\/span>
\n }<\/span><\/p>\n

\/\/ TokenStream \u7684\u57fa\u672c\u7ed3\u6784<\/span>
\n public<\/span> static<\/span> class<\/span> TokenStream<\/span><<\/span>T<\/span>><\/span><\/span> {<\/span>
\n private<\/span> final<\/span> Publisher<\/span><<\/span>T<\/span>><\/span><\/span> source;<\/span>
\n private<\/span> final<\/span> TokenProcessor<\/span><<\/span>T<\/span>><\/span><\/span> processor;<\/span>
\n private<\/span> final<\/span> FlowControl<\/span> flowControl;<\/span><\/p>\n

public<\/span> TokenStream<\/span>(<\/span>Publisher<\/span><<\/span>T<\/span>><\/span><\/span> source)<\/span> {<\/span>
\n this<\/span>.<\/span>source =<\/span> source;<\/span>
\n this<\/span>.<\/span>processor =<\/span> new<\/span> DefaultTokenProcessor<\/span><<\/span>><\/span><\/span>(<\/span>)<\/span>;<\/span>
\n this<\/span>.<\/span>flowControl =<\/span> new<\/span> RateBasedFlowControl<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

\/\/ \u6d41\u7684\u6838\u5fc3\u64cd\u4f5c<\/span>
\n public<\/span> void<\/span> subscribe<\/span>(<\/span>Subscriber<\/span><<\/span>T<\/span>><\/span><\/span> subscriber)<\/span> {<\/span>
\n source.<\/span>subscribe<\/span>(<\/span>new<\/span> SafeSubscriber<\/span><<\/span>><\/span><\/span>(<\/span>subscriber,<\/span> processor)<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ Token \u7684\u5b9a\u4e49<\/span>
\n public<\/span> static<\/span> class<\/span> Token<\/span> {<\/span>
\n private<\/span> final<\/span> String<\/span> content;<\/span>
\n private<\/span> final<\/span> int<\/span> index;<\/span>
\n private<\/span> final<\/span> boolean<\/span> isComplete;<\/span>
\n private<\/span> final<\/span> TokenType<\/span> type;<\/span>
\n private<\/span> final<\/span> Map<\/span><<\/span>String<\/span>,<\/span> Object<\/span>><\/span><\/span> metadata;<\/span><\/p>\n

public<\/span> enum<\/span> TokenType<\/span> {<\/span>
\n TEXT,<\/span> \/\/ \u6587\u672ctoken<\/span>
\n FUNCTION_CALL,<\/span> \/\/ \u51fd\u6570\u8c03\u7528<\/span>
\n TOOL_RESULT,<\/span> \/\/ \u5de5\u5177\u7ed3\u679c<\/span>
\n CONTROL,<\/span> \/\/ \u63a7\u5236token（\u5f00\u59cb\u3001\u7ed3\u675f\u3001\u6682\u505c）<\/span>
\n ERROR \/\/ \u9519\u8beftoken<\/span>
\n }<\/span>
\n }<\/span>
\n}<\/span><\/p>\n

1.2 TokenStream \u7684\u5b9e\u73b0\u6a21\u5f0f<\/h4>\n

\/**
\n * TokenStream \u7684\u591a\u79cd\u5b9e\u73b0\u6a21\u5f0f
\n *\/<\/span>
\npublic<\/span> class<\/span> TokenStreamImplementations<\/span> {<\/span><\/p>\n

\/\/ \u6a21\u5f0f1：\u54cd\u5e94\u5f0f\u6d41（Reactive Streams）<\/span>
\n public<\/span> static<\/span> class<\/span> ReactiveTokenStream<\/span><<\/span>T<\/span>><\/span><\/span> implements<\/span> Publisher<\/span><<\/span>T<\/span>><\/span><\/span> {<\/span>
\n private<\/span> final<\/span> List<\/span><<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span>><\/span><\/span> subscribers =<\/span>
\n new<\/span> CopyOnWriteArrayList<\/span><<\/span>><\/span><\/span>(<\/span>)<\/span>;<\/span>
\n private<\/span> volatile<\/span> boolean<\/span> cancelled =<\/span> false<\/span>;<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> subscribe<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber)<\/span> {<\/span>
\n if<\/span> (<\/span>cancelled)<\/span> {<\/span>
\n subscriber.<\/span>onError<\/span>(<\/span>new<\/span> IllegalStateException<\/span>(<\/span>"Stream cancelled"<\/span>)<\/span>)<\/span>;<\/span>
\n return<\/span>;<\/span>
\n }<\/span><\/p>\n

subscribers.<\/span>add<\/span>(<\/span>subscriber)<\/span>;<\/span>
\n subscriber.<\/span>onSubscribe<\/span>(<\/span>new<\/span> TokenSubscription<\/span>(<\/span>subscriber)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

private<\/span> class<\/span> TokenSubscription<\/span> implements<\/span> Subscription<\/span> {<\/span>
\n private<\/span> final<\/span> Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber;<\/span>
\n private<\/span> volatile<\/span> boolean<\/span> active =<\/span> true<\/span>;<\/span><\/p>\n

TokenSubscription<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber)<\/span> {<\/span>
\n this<\/span>.<\/span>subscriber =<\/span> subscriber;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> request<\/span>(<\/span>long<\/span> n)<\/span> {<\/span>
\n if<\/span> (<\/span>!<\/span>active ||<\/span> cancelled)<\/span> return<\/span>;<\/span><\/p>\n

\/\/ \u6a21\u62df\u751f\u6210token<\/span>
\n for<\/span> (<\/span>long<\/span> i =<\/span> 0<\/span>;<\/span> i <<\/span> n &&<\/span> active;<\/span> i++<\/span>)<\/span> {<\/span>
\n T<\/span> token =<\/span> generateNextToken<\/span>(<\/span>)<\/span>;<\/span>
\n if<\/span> (<\/span>token !=<\/span> null<\/span>)<\/span> {<\/span>
\n subscriber.<\/span>onNext<\/span>(<\/span>token)<\/span>;<\/span>
\n }<\/span> else<\/span> {<\/span>
\n subscriber.<\/span>onComplete<\/span>(<\/span>)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> cancel<\/span>(<\/span>)<\/span> {<\/span>
\n active =<\/span> false<\/span>;<\/span>
\n subscribers.<\/span>remove<\/span>(<\/span>subscriber)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u6a21\u5f0f2：\u56de\u8c03\u5f0f\u6d41（Callback-based）<\/span>
\n public<\/span> static<\/span> class<\/span> CallbackTokenStream<\/span><<\/span>T<\/span>><\/span><\/span> {<\/span>
\n private<\/span> final<\/span> TokenGenerator<\/span><<\/span>T<\/span>><\/span><\/span> generator;<\/span>
\n private<\/span> final<\/span> ExecutorService<\/span> executor;<\/span>
\n private<\/span> final<\/span> List<\/span><<\/span>TokenCallback<\/span><<\/span>T<\/span>><\/span>><\/span><\/span> callbacks =<\/span> new<\/span> ArrayList<\/span><<\/span>><\/span><\/span>(<\/span>)<\/span>;<\/span>
\n private<\/span> volatile<\/span> boolean<\/span> isStreaming =<\/span> false<\/span>;<\/span><\/p>\n

@FunctionalInterface<\/span>
\n public<\/span> interface<\/span> TokenCallback<\/span><<\/span>T<\/span>><\/span><\/span> {<\/span>
\n void<\/span> onToken<\/span>(<\/span>T<\/span> token)<\/span>;<\/span>
\n default<\/span> void<\/span> onError<\/span>(<\/span>Throwable<\/span> error)<\/span> {<\/span>}<\/span>
\n default<\/span> void<\/span> onComplete<\/span>(<\/span>)<\/span> {<\/span>}<\/span>
\n }<\/span><\/p>\n

public<\/span> void<\/span> startStreaming<\/span>(<\/span>TokenCallback<\/span><<\/span>T<\/span>><\/span><\/span> callback)<\/span> {<\/span>
\n if<\/span> (<\/span>isStreaming)<\/span> {<\/span>
\n throw<\/span> new<\/span> IllegalStateException<\/span>(<\/span>"Stream already in progress"<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

isStreaming =<\/span> true<\/span>;<\/span>
\n callbacks.<\/span>add<\/span>(<\/span>callback)<\/span>;<\/span><\/p>\n

executor.<\/span>submit<\/span>(<\/span>(<\/span>)<\/span> -><\/span> {<\/span>
\n try<\/span> {<\/span>
\n while<\/span> (<\/span>isStreaming)<\/span> {<\/span>
\n T<\/span> token =<\/span> generator.<\/span>generate<\/span>(<\/span>)<\/span>;<\/span>
\n if<\/span> (<\/span>token ==<\/span> null<\/span>)<\/span> {<\/span>
\n complete<\/span>(<\/span>)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span><\/p>\n

\/\/ \u540c\u6b65\u56de\u8c03\u901a\u77e5<\/span>
\n synchronized<\/span> (<\/span>callbacks)<\/span> {<\/span>
\n for<\/span> (<\/span>TokenCallback<\/span><<\/span>T<\/span>><\/span><\/span> cb :<\/span> callbacks)<\/span> {<\/span>
\n cb.<\/span>onToken<\/span>(<\/span>token)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span> catch<\/span> (<\/span>Exception<\/span> e)<\/span> {<\/span>
\n error<\/span>(<\/span>e)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

public<\/span> void<\/span> stopStreaming<\/span>(<\/span>)<\/span> {<\/span>
\n isStreaming =<\/span> false<\/span>;<\/span>
\n executor.<\/span>shutdown<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u6a21\u5f0f3：\u8fed\u4ee3\u5668\u5f0f\u6d41（Iterator-based）<\/span>
\n public<\/span> static<\/span> class<\/span> IteratorTokenStream<\/span><<\/span>T<\/span>><\/span><\/span> implements<\/span> Iterator<\/span><<\/span>T<\/span>><\/span><\/span>,<\/span> AutoCloseable<\/span> {<\/span>
\n private<\/span> final<\/span> TokenSource<\/span><<\/span>T<\/span>><\/span><\/span> source;<\/span>
\n private<\/span> T<\/span> nextToken;<\/span>
\n private<\/span> boolean<\/span> hasNext =<\/span> true<\/span>;<\/span>
\n private<\/span> volatile<\/span> boolean<\/span> closed =<\/span> false<\/span>;<\/span><\/p>\n

public<\/span> IteratorTokenStream<\/span>(<\/span>TokenSource<\/span><<\/span>T<\/span>><\/span><\/span> source)<\/span> {<\/span>
\n this<\/span>.<\/span>source =<\/span> source;<\/span>
\n fetchNext<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> fetchNext<\/span>(<\/span>)<\/span> {<\/span>
\n if<\/span> (<\/span>closed)<\/span> {<\/span>
\n hasNext =<\/span> false<\/span>;<\/span>
\n return<\/span>;<\/span>
\n }<\/span><\/p>\n

try<\/span> {<\/span>
\n nextToken =<\/span> source.<\/span>next<\/span>(<\/span>)<\/span>;<\/span>
\n hasNext =<\/span> (<\/span>nextToken !=<\/span> null<\/span>)<\/span>;<\/span>
\n }<\/span> catch<\/span> (<\/span>Exception<\/span> e)<\/span> {<\/span>
\n hasNext =<\/span> false<\/span>;<\/span>
\n throw<\/span> new<\/span> TokenStreamException<\/span>(<\/span>"Failed to fetch next token"<\/span>,<\/span> e)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> boolean<\/span> hasNext<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> hasNext &&<\/span> !<\/span>closed;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> T<\/span> next<\/span>(<\/span>)<\/span> {<\/span>
\n if<\/span> (<\/span>!<\/span>hasNext<\/span>(<\/span>)<\/span>)<\/span> {<\/span>
\n throw<\/span> new<\/span> NoSuchElementException<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

T<\/span> current =<\/span> nextToken;<\/span>
\n fetchNext<\/span>(<\/span>)<\/span>;<\/span>
\n return<\/span> current;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> close<\/span>(<\/span>)<\/span> {<\/span>
\n closed =<\/span> true<\/span>;<\/span>
\n source.<\/span>close<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n}<\/span><\/p>\n

\u4e8c\u3001\u6d41\u5f0f\u54cd\u5e94\u5f02\u5e38\u5904\u7406\u67b6\u6784<\/h3>\n

2.1 \u5f02\u5e38\u5206\u7c7b\u4e0e\u5904\u7406\u7b56\u7565<\/h4>\n

\/**
\n * \u6d41\u5f0f\u54cd\u5e94\u5f02\u5e38\u5206\u7c7b\u4f53\u7cfb
\n *\/<\/span>
\npublic<\/span> class<\/span> StreamExceptionHierarchy<\/span> {<\/span><\/p>\n

\/\/ \u5f02\u5e38\u57fa\u7c7b<\/span>
\n public<\/span> abstract<\/span> static<\/span> class<\/span> StreamException<\/span> extends<\/span> RuntimeException<\/span> {<\/span>
\n private<\/span> final<\/span> ErrorCategory<\/span> category;<\/span>
\n private<\/span> final<\/span> Instant<\/span> timestamp;<\/span>
\n private<\/span> final<\/span> String<\/span> streamId;<\/span><\/p>\n

public<\/span> StreamException<\/span>(<\/span>String<\/span> message,<\/span> ErrorCategory<\/span> category,<\/span>
\n String<\/span> streamId)<\/span> {<\/span>
\n super<\/span>(<\/span>message)<\/span>;<\/span>
\n this<\/span>.<\/span>category =<\/span> category;<\/span>
\n this<\/span>.<\/span>timestamp =<\/span> Instant<\/span>.<\/span>now<\/span>(<\/span>)<\/span>;<\/span>
\n this<\/span>.<\/span>streamId =<\/span> streamId;<\/span>
\n }<\/span><\/p>\n

public<\/span> abstract<\/span> boolean<\/span> isRecoverable<\/span>(<\/span>)<\/span>;<\/span>
\n public<\/span> abstract<\/span> RecoveryStrategy<\/span> getRecoveryStrategy<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

\/\/ \u9519\u8bef\u5206\u7c7b<\/span>
\n public<\/span> enum<\/span> ErrorCategory<\/span> {<\/span>
\n NETWORK,<\/span> \/\/ \u7f51\u7edc\u9519\u8bef<\/span>
\n RATE_LIMIT,<\/span> \/\/ \u901f\u7387\u9650\u5236<\/span>
\n TIMEOUT,<\/span> \/\/ \u8d85\u65f6<\/span>
\n MODEL,<\/span> \/\/ \u6a21\u578b\u9519\u8bef<\/span>
\n CLIENT,<\/span> \/\/ \u5ba2\u6237\u7aef\u9519\u8bef<\/span>
\n SERVER,<\/span> \/\/ \u670d\u52a1\u5668\u9519\u8bef<\/span>
\n VALIDATION,<\/span> \/\/ \u9a8c\u8bc1\u9519\u8bef<\/span>
\n INTERRUPTION \/\/ \u4e2d\u65ad\u9519\u8bef<\/span>
\n }<\/span><\/p>\n

\/\/ \u6062\u590d\u7b56\u7565<\/span>
\n public<\/span> enum<\/span> RecoveryStrategy<\/span> {<\/span>
\n RETRY,<\/span> \/\/ \u91cd\u8bd5<\/span>
\n RESUME,<\/span> \/\/ \u6062\u590d<\/span>
\n RESTART,<\/span> \/\/ \u91cd\u542f<\/span>
\n FALLBACK,<\/span> \/\/ \u964d\u7ea7<\/span>
\n FAIL_FAST,<\/span> \/\/ \u5feb\u901f\u5931\u8d25<\/span>
\n CONTINUE \/\/ \u7ee7\u7eed<\/span>
\n }<\/span><\/p>\n

\/\/ \u5177\u4f53\u5f02\u5e38\u7c7b\u578b<\/span>
\n public<\/span> static<\/span> class<\/span> NetworkException<\/span> extends<\/span> StreamException<\/span> {<\/span>
\n private<\/span> final<\/span> String<\/span> endpoint;<\/span>
\n private<\/span> final<\/span> int<\/span> statusCode;<\/span><\/p>\n

public<\/span> NetworkException<\/span>(<\/span>String<\/span> endpoint,<\/span> int<\/span> statusCode,<\/span>
\n String<\/span> streamId)<\/span> {<\/span>
\n super<\/span>(<\/span>String<\/span>.<\/span>format<\/span>(<\/span>"Network error connecting to %s (status: %d)"<\/span>,<\/span>
\n endpoint,<\/span> statusCode)<\/span>,<\/span>
\n ErrorCategory<\/span>.<\/span>NETWORK,<\/span> streamId)<\/span>;<\/span>
\n this<\/span>.<\/span>endpoint =<\/span> endpoint;<\/span>
\n this<\/span>.<\/span>statusCode =<\/span> statusCode;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> boolean<\/span> isRecoverable<\/span>(<\/span>)<\/span> {<\/span>
\n \/\/ 5xx\u9519\u8bef\u53ef\u80fd\u6062\u590d，4xx\u9519\u8bef\u4e0d\u53ef\u6062\u590d<\/span>
\n return<\/span> statusCode >=<\/span> 500<\/span> &&<\/span> statusCode <<\/span> 600<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> RecoveryStrategy<\/span> getRecoveryStrategy<\/span>(<\/span>)<\/span> {<\/span>
\n if<\/span> (<\/span>statusCode ==<\/span> 429<\/span>)<\/span> {<\/span> \/\/ \u901f\u7387\u9650\u5236<\/span>
\n return<\/span> RecoveryStrategy<\/span>.<\/span>RETRY;<\/span>
\n }<\/span> else<\/span> if<\/span> (<\/span>statusCode >=<\/span> 500<\/span>)<\/span> {<\/span>
\n return<\/span> RecoveryStrategy<\/span>.<\/span>RESUME;<\/span>
\n }<\/span> else<\/span> {<\/span>
\n return<\/span> RecoveryStrategy<\/span>.<\/span>FAIL_FAST;<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

public<\/span> static<\/span> class<\/span> RateLimitException<\/span> extends<\/span> StreamException<\/span> {<\/span>
\n private<\/span> final<\/span> Duration<\/span> retryAfter;<\/span>
\n private<\/span> final<\/span> int<\/span> limit;<\/span>
\n private<\/span> final<\/span> int<\/span> remaining;<\/span><\/p>\n

public<\/span> RateLimitException<\/span>(<\/span>Duration<\/span> retryAfter,<\/span> int<\/span> limit,<\/span>
\n int<\/span> remaining,<\/span> String<\/span> streamId)<\/span> {<\/span>
\n super<\/span>(<\/span>String<\/span>.<\/span>format<\/span>(<\/span>"Rate limit exceeded. Limit: %d, Remaining: %d"<\/span>,<\/span>
\n limit,<\/span> remaining)<\/span>,<\/span>
\n ErrorCategory<\/span>.<\/span>RATE_LIMIT,<\/span> streamId)<\/span>;<\/span>
\n this<\/span>.<\/span>retryAfter =<\/span> retryAfter;<\/span>
\n this<\/span>.<\/span>limit =<\/span> limit;<\/span>
\n this<\/span>.<\/span>remaining =<\/span> remaining;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> boolean<\/span> isRecoverable<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> true<\/span>;<\/span> \/\/ \u901f\u7387\u9650\u5236\u603b\u662f\u53ef\u6062\u590d\u7684<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> RecoveryStrategy<\/span> getRecoveryStrategy<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> RecoveryStrategy<\/span>.<\/span>RETRY;<\/span>
\n }<\/span><\/p>\n

public<\/span> Duration<\/span> getRetryAfter<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> retryAfter;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

public<\/span> static<\/span> class<\/span> StreamInterruptedException<\/span> extends<\/span> StreamException<\/span> {<\/span>
\n private<\/span> final<\/span> InterruptionSource<\/span> source;<\/span>
\n private<\/span> final<\/span> boolean<\/span> userInitiated;<\/span><\/p>\n

public<\/span> enum<\/span> InterruptionSource<\/span> {<\/span>
\n USER_CANCEL,<\/span> \/\/ \u7528\u6237\u53d6\u6d88<\/span>
\n TIMEOUT,<\/span> \/\/ \u8d85\u65f6<\/span>
\n RESOURCE_LIMIT,<\/span> \/\/ \u8d44\u6e90\u9650\u5236<\/span>
\n SYSTEM_SHUTDOWN,<\/span> \/\/ \u7cfb\u7edf\u5173\u95ed<\/span>
\n CONNECTION_LOST \/\/ \u8fde\u63a5\u4e22\u5931<\/span>
\n }<\/span><\/p>\n

public<\/span> StreamInterruptedException<\/span>(<\/span>InterruptionSource<\/span> source,<\/span>
\n boolean<\/span> userInitiated,<\/span>
\n String<\/span> streamId)<\/span> {<\/span>
\n super<\/span>(<\/span>String<\/span>.<\/span>format<\/span>(<\/span>"Stream interrupted by %s"<\/span>,<\/span> source)<\/span>,<\/span>
\n ErrorCategory<\/span>.<\/span>INTERRUPTION,<\/span> streamId)<\/span>;<\/span>
\n this<\/span>.<\/span>source =<\/span> source;<\/span>
\n this<\/span>.<\/span>userInitiated =<\/span> userInitiated;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> boolean<\/span> isRecoverable<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> source ==<\/span> InterruptionSource<\/span>.<\/span>CONNECTION_LOST;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> RecoveryStrategy<\/span> getRecoveryStrategy<\/span>(<\/span>)<\/span> {<\/span>
\n switch<\/span> (<\/span>source)<\/span> {<\/span>
\n case<\/span> USER_CANCEL:<\/span>
\n return<\/span> RecoveryStrategy<\/span>.<\/span>FAIL_FAST;<\/span>
\n case<\/span> TIMEOUT:<\/span>
\n return<\/span> RecoveryStrategy<\/span>.<\/span>RESUME;<\/span>
\n case<\/span> CONNECTION_LOST:<\/span>
\n return<\/span> RecoveryStrategy<\/span>.<\/span>RESTART;<\/span>
\n default<\/span>:<\/span>
\n return<\/span> RecoveryStrategy<\/span>.<\/span>FAIL_FAST;<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span>
\n}<\/span><\/p>\n

2.2 \u5f02\u5e38\u5904\u7406\u6846\u67b6\u5b9e\u73b0<\/h4>\n

\/**
\n * \u5b8c\u6574\u7684\u5f02\u5e38\u5904\u7406\u6846\u67b6
\n *\/<\/span>
\npublic<\/span> class<\/span> StreamExceptionHandler<\/span> {<\/span><\/p>\n

\/\/ \u5f02\u5e38\u5904\u7406\u5668\u63a5\u53e3<\/span>
\n @FunctionalInterface<\/span>
\n public<\/span> interface<\/span> ExceptionHandler<\/span><<\/span>T<\/span> extends<\/span> StreamException<\/span>><\/span><\/span> {<\/span>
\n HandlingResult<\/span> handle<\/span>(<\/span>T<\/span> exception,<\/span> StreamContext<\/span> context)<\/span>;<\/span>
\n }<\/span><\/p>\n

\/\/ \u5904\u7406\u7ed3\u679c<\/span>
\n public<\/span> static<\/span> class<\/span> HandlingResult<\/span> {<\/span>
\n private<\/span> final<\/span> boolean<\/span> shouldContinue;<\/span>
\n private<\/span> final<\/span> RecoveryAction<\/span> recoveryAction;<\/span>
\n private<\/span> final<\/span> String<\/span> message;<\/span><\/p>\n

public<\/span> enum<\/span> RecoveryAction<\/span> {<\/span>
\n NONE,<\/span>
\n RETRY_IMMEDIATELY,<\/span>
\n RETRY_AFTER_DELAY,<\/span>
\n SWITCH_ENDPOINT,<\/span>
\n USE_FALLBACK,<\/span>
\n RECONNECT_STREAM
\n }<\/span><\/p>\n

public<\/span> HandlingResult<\/span>(<\/span>boolean<\/span> shouldContinue,<\/span>
\n RecoveryAction<\/span> recoveryAction,<\/span>
\n String<\/span> message)<\/span> {<\/span>
\n this<\/span>.<\/span>shouldContinue =<\/span> shouldContinue;<\/span>
\n this<\/span>.<\/span>recoveryAction =<\/span> recoveryAction;<\/span>
\n this<\/span>.<\/span>message =<\/span> message;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u6d41\u4e0a\u4e0b\u6587<\/span>
\n public<\/span> static<\/span> class<\/span> StreamContext<\/span> {<\/span>
\n private<\/span> final<\/span> String<\/span> streamId;<\/span>
\n private<\/span> final<\/span> Instant<\/span> startTime;<\/span>
\n private<\/span> volatile<\/span> Instant<\/span> lastTokenTime;<\/span>
\n private<\/span> final<\/span> AtomicInteger<\/span> tokenCount;<\/span>
\n private<\/span> final<\/span> Map<\/span><<\/span>String<\/span>,<\/span> Object<\/span>><\/span><\/span> metadata;<\/span><\/p>\n

public<\/span> StreamContext<\/span>(<\/span>String<\/span> streamId)<\/span> {<\/span>
\n this<\/span>.<\/span>streamId =<\/span> streamId;<\/span>
\n this<\/span>.<\/span>startTime =<\/span> Instant<\/span>.<\/span>now<\/span>(<\/span>)<\/span>;<\/span>
\n this<\/span>.<\/span>lastTokenTime =<\/span> startTime;<\/span>
\n this<\/span>.<\/span>tokenCount =<\/span> new<\/span> AtomicInteger<\/span>(<\/span>0<\/span>)<\/span>;<\/span>
\n this<\/span>.<\/span>metadata =<\/span> new<\/span> ConcurrentHashMap<\/span><<\/span>><\/span><\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

public<\/span> void<\/span> recordToken<\/span>(<\/span>)<\/span> {<\/span>
\n tokenCount.<\/span>incrementAndGet<\/span>(<\/span>)<\/span>;<\/span>
\n lastTokenTime =<\/span> Instant<\/span>.<\/span>now<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u5f02\u5e38\u5904\u7406\u5668\u6ce8\u518c\u4e2d\u5fc3<\/span>
\n public<\/span> static<\/span> class<\/span> ExceptionHandlerRegistry<\/span> {<\/span>
\n private<\/span> final<\/span> Map<\/span><<\/span>Class<\/span><<\/span>?<\/span> extends<\/span> StreamException<\/span>><\/span>,<\/span>
\n ExceptionHandler<\/span><<\/span>?<\/span>><\/span>><\/span><\/span> handlers =<\/span> new<\/span> ConcurrentHashMap<\/span><<\/span>><\/span><\/span>(<\/span>)<\/span>;<\/span>
\n private<\/span> final<\/span> ExceptionHandler<\/span><<\/span>StreamException<\/span>><\/span><\/span> defaultHandler;<\/span><\/p>\n

public<\/span> ExceptionHandlerRegistry<\/span>(<\/span>)<\/span> {<\/span>
\n this<\/span>.<\/span>defaultHandler =<\/span> (<\/span>ex,<\/span> ctx)<\/span> -><\/span>
\n new<\/span> HandlingResult<\/span>(<\/span>false<\/span>,<\/span>
\n HandlingResult.<\/span>RecoveryAction<\/span>.<\/span>NONE,<\/span>
\n "Unhandled exception: "<\/span> +<\/span> ex.<\/span>getMessage<\/span>(<\/span>)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

public<\/span> <<\/span>T<\/span> extends<\/span> StreamException<\/span>><\/span><\/span> void<\/span> registerHandler<\/span>(<\/span>
\n Class<\/span><<\/span>T<\/span>><\/span><\/span> exceptionType,<\/span>
\n ExceptionHandler<\/span><<\/span>T<\/span>><\/span><\/span> handler)<\/span> {<\/span>
\n handlers.<\/span>put<\/span>(<\/span>exceptionType,<\/span> handler)<\/span>;<\/span>
\n }<\/span><\/p>\n

@SuppressWarnings<\/span>(<\/span>"unchecked"<\/span>)<\/span>
\n public<\/span> HandlingResult<\/span> handle<\/span>(<\/span>StreamException<\/span> exception,<\/span>
\n StreamContext<\/span> context)<\/span> {<\/span>
\n ExceptionHandler<\/span><<\/span>StreamException<\/span>><\/span><\/span> handler =<\/span>
\n (<\/span>ExceptionHandler<\/span><<\/span>StreamException<\/span>><\/span><\/span>)<\/span>
\n handlers.<\/span>get<\/span>(<\/span>exception.<\/span>getClass<\/span>(<\/span>)<\/span>)<\/span>;<\/span><\/p>\n

if<\/span> (<\/span>handler ==<\/span> null<\/span>)<\/span> {<\/span>
\n \/\/ \u67e5\u627e\u7236\u7c7b\u5904\u7406\u5668<\/span>
\n for<\/span> (<\/span>Map.<\/span>Entry<\/span><<\/span>Class<\/span><<\/span>?<\/span> extends<\/span> StreamException<\/span>><\/span>,<\/span>
\n ExceptionHandler<\/span><<\/span>?<\/span>><\/span>><\/span><\/span> entry :<\/span> handlers.<\/span>entrySet<\/span>(<\/span>)<\/span>)<\/span> {<\/span>
\n if<\/span> (<\/span>entry.<\/span>getKey<\/span>(<\/span>)<\/span>.<\/span>isAssignableFrom<\/span>(<\/span>exception.<\/span>getClass<\/span>(<\/span>)<\/span>)<\/span>)<\/span> {<\/span>
\n handler =<\/span> (<\/span>ExceptionHandler<\/span><<\/span>StreamException<\/span>><\/span><\/span>)<\/span> entry.<\/span>getValue<\/span>(<\/span>)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

if<\/span> (<\/span>handler ==<\/span> null<\/span>)<\/span> {<\/span>
\n handler =<\/span> defaultHandler;<\/span>
\n }<\/span><\/p>\n

return<\/span> handler.<\/span>handle<\/span>(<\/span>exception,<\/span> context)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u5177\u4f53\u5904\u7406\u5668\u5b9e\u73b0<\/span>
\n public<\/span> static<\/span> class<\/span> NetworkExceptionHandler<\/span>
\n implements<\/span> ExceptionHandler<\/span><<\/span>NetworkException<\/span>><\/span><\/span> {<\/span><\/p>\n

private<\/span> final<\/span> RetryPolicy<\/span> retryPolicy;<\/span>
\n private<\/span> final<\/span> CircuitBreaker<\/span> circuitBreaker;<\/span>
\n private<\/span> final<\/span> EndpointManager<\/span> endpointManager;<\/span><\/p>\n

public<\/span> NetworkExceptionHandler<\/span>(<\/span>RetryPolicy<\/span> retryPolicy,<\/span>
\n CircuitBreaker<\/span> circuitBreaker,<\/span>
\n EndpointManager<\/span> endpointManager)<\/span> {<\/span>
\n this<\/span>.<\/span>retryPolicy =<\/span> retryPolicy;<\/span>
\n this<\/span>.<\/span>circuitBreaker =<\/span> circuitBreaker;<\/span>
\n this<\/span>.<\/span>endpointManager =<\/span> endpointManager;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> HandlingResult<\/span> handle<\/span>(<\/span>NetworkException<\/span> exception,<\/span>
\n StreamContext<\/span> context)<\/span> {<\/span><\/p>\n

if<\/span> (<\/span>!<\/span>exception.<\/span>isRecoverable<\/span>(<\/span>)<\/span>)<\/span> {<\/span>
\n circuitBreaker.<\/span>recordFailure<\/span>(<\/span>exception.<\/span>getEndpoint<\/span>(<\/span>)<\/span>)<\/span>;<\/span>
\n return<\/span> new<\/span> HandlingResult<\/span>(<\/span>false<\/span>,<\/span>
\n HandlingResult.<\/span>RecoveryAction<\/span>.<\/span>NONE,<\/span>
\n "Non-recoverable network error"<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

\/\/ \u68c0\u67e5\u662f\u5426\u5e94\u8be5\u91cd\u8bd5<\/span>
\n if<\/span> (<\/span>retryPolicy.<\/span>shouldRetry<\/span>(<\/span>exception,<\/span> context)<\/span>)<\/span> {<\/span>
\n Duration<\/span> delay =<\/span> retryPolicy.<\/span>getNextRetryDelay<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u68c0\u67e5\u662f\u5426\u5e94\u8be5\u5207\u6362\u7aef\u70b9<\/span>
\n if<\/span> (<\/span>endpointManager.<\/span>shouldSwitchEndpoint<\/span>(<\/span>
\n exception.<\/span>getEndpoint<\/span>(<\/span>)<\/span>,<\/span> exception.<\/span>getStatusCode<\/span>(<\/span>)<\/span>)<\/span>)<\/span> {<\/span><\/p>\n

String<\/span> newEndpoint =<\/span> endpointManager.<\/span>getNextEndpoint<\/span>(<\/span>)<\/span>;<\/span>
\n return<\/span> new<\/span> HandlingResult<\/span>(<\/span>true<\/span>,<\/span>
\n HandlingResult.<\/span>RecoveryAction<\/span>.<\/span>SWITCH_ENDPOINT,<\/span>
\n String<\/span>.<\/span>format<\/span>(<\/span>"Switching to endpoint: %s after delay: %s"<\/span>,<\/span>
\n newEndpoint,<\/span> delay)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

return<\/span> new<\/span> HandlingResult<\/span>(<\/span>true<\/span>,<\/span>
\n HandlingResult.<\/span>RecoveryAction<\/span>.<\/span>RETRY_AFTER_DELAY,<\/span>
\n String<\/span>.<\/span>format<\/span>(<\/span>"Will retry after %s"<\/span>,<\/span> delay)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

return<\/span> new<\/span> HandlingResult<\/span>(<\/span>false<\/span>,<\/span>
\n HandlingResult.<\/span>RecoveryAction<\/span>.<\/span>USE_FALLBACK,<\/span>
\n "Retry limit exceeded, using fallback"<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

public<\/span> static<\/span> class<\/span> RateLimitExceptionHandler<\/span>
\n implements<\/span> ExceptionHandler<\/span><<\/span>RateLimitException<\/span>><\/span><\/span> {<\/span><\/p>\n

@Override<\/span>
\n public<\/span> HandlingResult<\/span> handle<\/span>(<\/span>RateLimitException<\/span> exception,<\/span>
\n StreamContext<\/span> context)<\/span> {<\/span><\/p>\n

Duration<\/span> retryAfter =<\/span> exception.<\/span>getRetryAfter<\/span>(<\/span>)<\/span>;<\/span>
\n if<\/span> (<\/span>retryAfter ==<\/span> null<\/span>)<\/span> {<\/span>
\n \/\/ \u4f7f\u7528\u6307\u6570\u9000\u907f<\/span>
\n retryAfter =<\/span> calculateExponentialBackoff<\/span>(<\/span>context)<\/span>;<\/span>
\n }<\/span><\/p>\n

return<\/span> new<\/span> HandlingResult<\/span>(<\/span>true<\/span>,<\/span>
\n HandlingResult.<\/span>RecoveryAction<\/span>.<\/span>RETRY_AFTER_DELAY,<\/span>
\n String<\/span>.<\/span>format<\/span>(<\/span>"Rate limited, retrying after %s"<\/span>,<\/span> retryAfter)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

private<\/span> Duration<\/span> calculateExponentialBackoff<\/span>(<\/span>StreamContext<\/span> context)<\/span> {<\/span>
\n \/\/ \u7b80\u5355\u7684\u6307\u6570\u9000\u907f\u7b97\u6cd5<\/span>
\n int<\/span> attemptCount =<\/span> getAttemptCount<\/span>(<\/span>context)<\/span>;<\/span>
\n long<\/span> delayMillis =<\/span> Math<\/span>.<\/span>min<\/span>(<\/span>
\n 1000L<\/span> *<\/span> (<\/span>long<\/span>)<\/span> Math<\/span>.<\/span>pow<\/span>(<\/span>2<\/span>,<\/span> attemptCount)<\/span>,<\/span>
\n 60000L<\/span> \/\/ \u6700\u59271\u5206\u949f<\/span>
\n )<\/span>;<\/span>
\n return<\/span> Duration<\/span>.<\/span>ofMillis<\/span>(<\/span>delayMillis)<\/span>;<\/span>
\n }<\/span><\/p>\n

private<\/span> int<\/span> getAttemptCount<\/span>(<\/span>StreamContext<\/span> context)<\/span> {<\/span>
\n \/\/ \u4ece\u4e0a\u4e0b\u6587\u83b7\u53d6\u5c1d\u8bd5\u6b21\u6570<\/span>
\n return<\/span> (<\/span>int<\/span>)<\/span> context.<\/span>getMetadata<\/span>(<\/span>)<\/span>
\n .<\/span>getOrDefault<\/span>(<\/span>"rateLimitAttempts"<\/span>,<\/span> 0<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

public<\/span> static<\/span> class<\/span> InterruptionHandler<\/span>
\n implements<\/span> ExceptionHandler<\/span><<\/span>StreamInterruptedException<\/span>><\/span><\/span> {<\/span><\/p>\n

private<\/span> final<\/span> StreamRecoveryManager<\/span> recoveryManager;<\/span>
\n private<\/span> final<\/span> boolean<\/span> allowResume;<\/span><\/p>\n

public<\/span> InterruptionHandler<\/span>(<\/span>StreamRecoveryManager<\/span> recoveryManager,<\/span>
\n boolean<\/span> allowResume)<\/span> {<\/span>
\n this<\/span>.<\/span>recoveryManager =<\/span> recoveryManager;<\/span>
\n this<\/span>.<\/span>allowResume =<\/span> allowResume;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> HandlingResult<\/span> handle<\/span>(<\/span>StreamInterruptedException<\/span> exception,<\/span>
\n StreamContext<\/span> context)<\/span> {<\/span><\/p>\n

if<\/span> (<\/span>!<\/span>exception.<\/span>isRecoverable<\/span>(<\/span>)<\/span>)<\/span> {<\/span>
\n return<\/span> new<\/span> HandlingResult<\/span>(<\/span>false<\/span>,<\/span>
\n HandlingResult.<\/span>RecoveryAction<\/span>.<\/span>NONE,<\/span>
\n "Non-recoverable interruption"<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

if<\/span> (<\/span>allowResume &&<\/span> recoveryManager.<\/span>canResume<\/span>(<\/span>context)<\/span>)<\/span> {<\/span>
\n \/\/ \u5c1d\u8bd5\u6062\u590d\u6d41<\/span>
\n boolean<\/span> resumed =<\/span> recoveryManager.<\/span>resumeStream<\/span>(<\/span>context)<\/span>;<\/span>
\n if<\/span> (<\/span>resumed)<\/span> {<\/span>
\n return<\/span> new<\/span> HandlingResult<\/span>(<\/span>true<\/span>,<\/span>
\n HandlingResult.<\/span>RecoveryAction<\/span>.<\/span>RECONNECT_STREAM,<\/span>
\n "Stream resumed successfully"<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u5982\u679c\u65e0\u6cd5\u6062\u590d，\u5c1d\u8bd5\u91cd\u542f<\/span>
\n boolean<\/span> restarted =<\/span> recoveryManager.<\/span>restartStream<\/span>(<\/span>context)<\/span>;<\/span>
\n if<\/span> (<\/span>restarted)<\/span> {<\/span>
\n return<\/span> new<\/span> HandlingResult<\/span>(<\/span>true<\/span>,<\/span>
\n HandlingResult.<\/span>RecoveryAction<\/span>.<\/span>RESTART,<\/span>
\n "Stream restarted"<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

return<\/span> new<\/span> HandlingResult<\/span>(<\/span>false<\/span>,<\/span>
\n HandlingResult.<\/span>RecoveryAction<\/span>.<\/span>USE_FALLBACK,<\/span>
\n "Unable to recover stream, using fallback"<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n}<\/span><\/p>\n

\u4e09\u3001\u6d41\u5f0f\u54cd\u5e94\u5b8c\u6574\u5b9e\u73b0<\/h3>\n

3.1 \u5b89\u5168\u7684 TokenStream \u5b9e\u73b0<\/h4>\n

\/**
\n * \u5e26\u5b8c\u6574\u5f02\u5e38\u5904\u7406\u7684 TokenStream \u5b9e\u73b0
\n *\/<\/span>
\npublic<\/span> class<\/span> SafeTokenStream<\/span><<\/span>T<\/span>><\/span><\/span> implements<\/span> Publisher<\/span><<\/span>T<\/span>><\/span><\/span>,<\/span> AutoCloseable<\/span> {<\/span><\/p>\n

private<\/span> final<\/span> TokenSource<\/span><<\/span>T<\/span>><\/span><\/span> tokenSource;<\/span>
\n private<\/span> final<\/span> ExecutorService<\/span> executor;<\/span>
\n private<\/span> final<\/span> ExceptionHandlerRegistry<\/span> exceptionHandler;<\/span>
\n private<\/span> final<\/span> StreamRecoveryManager<\/span> recoveryManager;<\/span>
\n private<\/span> final<\/span> CircuitBreaker<\/span> circuitBreaker;<\/span>
\n private<\/span> final<\/span> RateLimiter<\/span> rateLimiter;<\/span><\/p>\n

private<\/span> final<\/span> List<\/span><<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span>><\/span><\/span> subscribers =<\/span>
\n new<\/span> CopyOnWriteArrayList<\/span><<\/span>><\/span><\/span>(<\/span>)<\/span>;<\/span>
\n private<\/span> final<\/span> Map<\/span><<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span>,<\/span> StreamContext<\/span>><\/span><\/span> contexts =<\/span>
\n new<\/span> ConcurrentHashMap<\/span><<\/span>><\/span><\/span>(<\/span>)<\/span>;<\/span><\/p>\n

private<\/span> volatile<\/span> boolean<\/span> isClosed =<\/span> false<\/span>;<\/span>
\n private<\/span> final<\/span> AtomicInteger<\/span> activeStreams =<\/span> new<\/span> AtomicInteger<\/span>(<\/span>0<\/span>)<\/span>;<\/span><\/p>\n

public<\/span> SafeTokenStream<\/span>(<\/span>TokenSource<\/span><<\/span>T<\/span>><\/span><\/span> tokenSource,<\/span>
\n ExecutorService<\/span> executor,<\/span>
\n ExceptionHandlerRegistry<\/span> exceptionHandler,<\/span>
\n StreamRecoveryManager<\/span> recoveryManager)<\/span> {<\/span>
\n this<\/span>.<\/span>tokenSource =<\/span> tokenSource;<\/span>
\n this<\/span>.<\/span>executor =<\/span> executor;<\/span>
\n this<\/span>.<\/span>exceptionHandler =<\/span> exceptionHandler;<\/span>
\n this<\/span>.<\/span>recoveryManager =<\/span> recoveryManager;<\/span>
\n this<\/span>.<\/span>circuitBreaker =<\/span> new<\/span> CircuitBreaker<\/span>(<\/span>5<\/span>,<\/span> Duration<\/span>.<\/span>ofMinutes<\/span>(<\/span>1<\/span>)<\/span>)<\/span>;<\/span>
\n this<\/span>.<\/span>rateLimiter =<\/span> RateLimiter<\/span>.<\/span>create<\/span>(<\/span>100.0<\/span>)<\/span>;<\/span> \/\/ \u6bcf\u79d2100\u4e2atoken<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> subscribe<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber)<\/span> {<\/span>
\n if<\/span> (<\/span>isClosed)<\/span> {<\/span>
\n subscriber.<\/span>onError<\/span>(<\/span>new<\/span> IllegalStateException<\/span>(<\/span>"Stream is closed"<\/span>)<\/span>)<\/span>;<\/span>
\n return<\/span>;<\/span>
\n }<\/span><\/p>\n

if<\/span> (<\/span>subscribers.<\/span>contains<\/span>(<\/span>subscriber)<\/span>)<\/span> {<\/span>
\n subscriber.<\/span>onError<\/span>(<\/span>new<\/span> IllegalStateException<\/span>(<\/span>"Already subscribed"<\/span>)<\/span>)<\/span>;<\/span>
\n return<\/span>;<\/span>
\n }<\/span><\/p>\n

subscribers.<\/span>add<\/span>(<\/span>subscriber)<\/span>;<\/span>
\n StreamContext<\/span> context =<\/span> new<\/span> StreamContext<\/span>(<\/span>generateStreamId<\/span>(<\/span>)<\/span>)<\/span>;<\/span>
\n contexts.<\/span>put<\/span>(<\/span>subscriber,<\/span> context)<\/span>;<\/span><\/p>\n

subscriber.<\/span>onSubscribe<\/span>(<\/span>new<\/span> SafeSubscription<\/span>(<\/span>subscriber,<\/span> context)<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u5f02\u6b65\u542f\u52a8\u6d41\u5904\u7406<\/span>
\n executor.<\/span>submit<\/span>(<\/span>(<\/span>)<\/span> -><\/span> processStream<\/span>(<\/span>subscriber,<\/span> context)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

private<\/span> class<\/span> SafeSubscription<\/span> implements<\/span> Subscription<\/span> {<\/span>
\n private<\/span> final<\/span> Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber;<\/span>
\n private<\/span> final<\/span> StreamContext<\/span> context;<\/span>
\n private<\/span> volatile<\/span> boolean<\/span> cancelled =<\/span> false<\/span>;<\/span>
\n private<\/span> final<\/span> AtomicLong<\/span> requested =<\/span> new<\/span> AtomicLong<\/span>(<\/span>0<\/span>)<\/span>;<\/span><\/p>\n

public<\/span> SafeSubscription<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber,<\/span>
\n StreamContext<\/span> context)<\/span> {<\/span>
\n this<\/span>.<\/span>subscriber =<\/span> subscriber;<\/span>
\n this<\/span>.<\/span>context =<\/span> context;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> request<\/span>(<\/span>long<\/span> n)<\/span> {<\/span>
\n if<\/span> (<\/span>n <=<\/span> 0<\/span>)<\/span> {<\/span>
\n subscriber.<\/span>onError<\/span>(<\/span>new<\/span> IllegalArgumentException<\/span>(<\/span>
\n "Request must be positive"<\/span>)<\/span>)<\/span>;<\/span>
\n return<\/span>;<\/span>
\n }<\/span><\/p>\n

requested.<\/span>addAndGet<\/span>(<\/span>n)<\/span>;<\/span>
\n synchronized<\/span> (<\/span>this<\/span>)<\/span> {<\/span>
\n notify<\/span>(<\/span>)<\/span>;<\/span> \/\/ \u5524\u9192\u7b49\u5f85\u7684\u6d41\u5904\u7406\u7ebf\u7a0b<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> cancel<\/span>(<\/span>)<\/span> {<\/span>
\n cancelled =<\/span> true<\/span>;<\/span>
\n cleanup<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

public<\/span> boolean<\/span> isCancelled<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> cancelled;<\/span>
\n }<\/span><\/p>\n

public<\/span> long<\/span> getRequested<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> requested.<\/span>get<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

public<\/span> void<\/span> consumeRequest<\/span>(<\/span>)<\/span> {<\/span>
\n requested.<\/span>decrementAndGet<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> processStream<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber,<\/span>
\n StreamContext<\/span> context)<\/span> {<\/span><\/p>\n

activeStreams.<\/span>incrementAndGet<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

try<\/span> {<\/span>
\n while<\/span> (<\/span>!<\/span>isClosed &&<\/span> !<\/span>getSubscription<\/span>(<\/span>subscriber)<\/span>.<\/span>isCancelled<\/span>(<\/span>)<\/span>)<\/span> {<\/span><\/p>\n

\/\/ \u68c0\u67e5\u662f\u5426\u6709\u8bf7\u6c42\u7684token<\/span>
\n SafeSubscription<\/span> subscription =<\/span> getSubscription<\/span>(<\/span>subscriber)<\/span>;<\/span>
\n if<\/span> (<\/span>subscription.<\/span>getRequested<\/span>(<\/span>)<\/span> <=<\/span> 0<\/span>)<\/span> {<\/span>
\n synchronized<\/span> (<\/span>subscription)<\/span> {<\/span>
\n try<\/span> {<\/span>
\n subscription.<\/span>wait<\/span>(<\/span>1000<\/span>)<\/span>;<\/span> \/\/ \u7b49\u5f85\u8bf7\u6c42<\/span>
\n }<\/span> catch<\/span> (<\/span>InterruptedException<\/span> e)<\/span> {<\/span>
\n Thread<\/span>.<\/span>currentThread<\/span>(<\/span>)<\/span>.<\/span>interrupt<\/span>(<\/span>)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n continue<\/span>;<\/span>
\n }<\/span><\/p>\n

\/\/ \u9650\u6d41\u63a7\u5236<\/span>
\n rateLimiter.<\/span>acquire<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u65ad\u8def\u5668\u68c0\u67e5<\/span>
\n if<\/span> (<\/span>!<\/span>circuitBreaker.<\/span>allowRequest<\/span>(<\/span>)<\/span>)<\/span> {<\/span>
\n handleException<\/span>(<\/span>new<\/span> CircuitBreakerOpenException<\/span>(<\/span>
\n "Circuit breaker is open"<\/span>,<\/span> context.<\/span>getStreamId<\/span>(<\/span>)<\/span>)<\/span>,<\/span>
\n subscriber,<\/span> context)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span><\/p>\n

try<\/span> {<\/span>
\n \/\/ \u751f\u6210\u4e0b\u4e00\u4e2atoken<\/span>
\n T<\/span> token =<\/span> tokenSource.<\/span>nextToken<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

if<\/span> (<\/span>token ==<\/span> null<\/span>)<\/span> {<\/span>
\n \/\/ \u6d41\u7ed3\u675f<\/span>
\n subscriber.<\/span>onComplete<\/span>(<\/span>)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span><\/p>\n

\/\/ \u53d1\u9001token<\/span>
\n subscriber.<\/span>onNext<\/span>(<\/span>token)<\/span>;<\/span>
\n subscription.<\/span>consumeRequest<\/span>(<\/span>)<\/span>;<\/span>
\n context.<\/span>recordToken<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

}<\/span> catch<\/span> (<\/span>StreamInterruptedException<\/span> e)<\/span> {<\/span>
\n \/\/ \u5904\u7406\u4e2d\u65ad<\/span>
\n HandlingResult<\/span> result =<\/span> exceptionHandler.<\/span>handle<\/span>(<\/span>e,<\/span> context)<\/span>;<\/span><\/p>\n

if<\/span> (<\/span>!<\/span>result.<\/span>shouldContinue<\/span>(<\/span>)<\/span>)<\/span> {<\/span>
\n handleException<\/span>(<\/span>e,<\/span> subscriber,<\/span> context)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span><\/p>\n

\/\/ \u6839\u636e\u6062\u590d\u7b56\u7565\u5904\u7406<\/span>
\n handleRecovery<\/span>(<\/span>result,<\/span> subscriber,<\/span> context)<\/span>;<\/span><\/p>\n

}<\/span> catch<\/span> (<\/span>StreamException<\/span> e)<\/span> {<\/span>
\n \/\/ \u5904\u7406\u5176\u4ed6\u6d41\u5f02\u5e38<\/span>
\n HandlingResult<\/span> result =<\/span> exceptionHandler.<\/span>handle<\/span>(<\/span>e,<\/span> context)<\/span>;<\/span><\/p>\n

if<\/span> (<\/span>!<\/span>result.<\/span>shouldContinue<\/span>(<\/span>)<\/span>)<\/span> {<\/span>
\n handleException<\/span>(<\/span>e,<\/span> subscriber,<\/span> context)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span><\/p>\n

handleRecovery<\/span>(<\/span>result,<\/span> subscriber,<\/span> context)<\/span>;<\/span><\/p>\n

}<\/span> catch<\/span> (<\/span>Exception<\/span> e)<\/span> {<\/span>
\n \/\/ \u5904\u7406\u975e\u6d41\u5f02\u5e38<\/span>
\n StreamException<\/span> wrapped =<\/span> new<\/span> GenericStreamException<\/span>(<\/span>
\n "Unexpected error"<\/span>,<\/span> e,<\/span> context.<\/span>getStreamId<\/span>(<\/span>)<\/span>)<\/span>;<\/span>
\n handleException<\/span>(<\/span>wrapped,<\/span> subscriber,<\/span> context)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span> finally<\/span> {<\/span>
\n cleanupSubscriber<\/span>(<\/span>subscriber)<\/span>;<\/span>
\n activeStreams.<\/span>decrementAndGet<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> handleException<\/span>(<\/span>StreamException<\/span> exception,<\/span>
\n Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber,<\/span>
\n StreamContext<\/span> context)<\/span> {<\/span><\/p>\n

\/\/ \u8bb0\u5f55\u5f02\u5e38<\/span>
\n context.<\/span>getMetadata<\/span>(<\/span>)<\/span>.<\/span>put<\/span>(<\/span>"lastError"<\/span>,<\/span> exception)<\/span>;<\/span><\/p>\n

\/\/ \u901a\u77e5\u8ba2\u9605\u8005<\/span>
\n subscriber.<\/span>onError<\/span>(<\/span>exception)<\/span>;<\/span><\/p>\n

\/\/ \u6e05\u7406\u8d44\u6e90<\/span>
\n cleanupSubscriber<\/span>(<\/span>subscriber)<\/span>;<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> handleRecovery<\/span>(<\/span>HandlingResult<\/span> result,<\/span>
\n Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber,<\/span>
\n StreamContext<\/span> context)<\/span> {<\/span><\/p>\n

switch<\/span> (<\/span>result.<\/span>getRecoveryAction<\/span>(<\/span>)<\/span>)<\/span> {<\/span>
\n case<\/span> RETRY_IMMEDIATELY:<\/span>
\n \/\/ \u7acb\u5373\u91cd\u8bd5，\u65e0\u9700\u5ef6\u8fdf<\/span>
\n break<\/span>;<\/span><\/p>\n

case<\/span> RETRY_AFTER_DELAY:<\/span>
\n \/\/ \u5ef6\u8fdf\u540e\u91cd\u8bd5<\/span>
\n try<\/span> {<\/span>
\n Thread<\/span>.<\/span>sleep<\/span>(<\/span>extractDelay<\/span>(<\/span>result)<\/span>)<\/span>;<\/span>
\n }<\/span> catch<\/span> (<\/span>InterruptedException<\/span> e)<\/span> {<\/span>
\n Thread<\/span>.<\/span>currentThread<\/span>(<\/span>)<\/span>.<\/span>interrupt<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n break<\/span>;<\/span><\/p>\n

case<\/span> SWITCH_ENDPOINT:<\/span>
\n \/\/ \u5207\u6362\u7aef\u70b9<\/span>
\n tokenSource.<\/span>switchEndpoint<\/span>(<\/span>)<\/span>;<\/span>
\n break<\/span>;<\/span><\/p>\n

case<\/span> RECONNECT_STREAM:<\/span>
\n \/\/ \u91cd\u65b0\u8fde\u63a5\u6d41<\/span>
\n recoveryManager.<\/span>reconnect<\/span>(<\/span>context)<\/span>;<\/span>
\n break<\/span>;<\/span><\/p>\n

case<\/span> USE_FALLBACK:<\/span>
\n \/\/ \u4f7f\u7528\u964d\u7ea7\u7b56\u7565<\/span>
\n useFallback<\/span>(<\/span>subscriber,<\/span> context)<\/span>;<\/span>
\n break<\/span>;<\/span><\/p>\n

case<\/span> RESTART:<\/span>
\n \/\/ \u91cd\u542f\u6d41<\/span>
\n restartStream<\/span>(<\/span>subscriber,<\/span> context)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> useFallback<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber,<\/span>
\n StreamContext<\/span> context)<\/span> {<\/span>
\n \/\/ \u5b9e\u73b0\u964d\u7ea7\u903b\u8f91<\/span>
\n try<\/span> {<\/span>
\n T<\/span> fallbackToken =<\/span> getFallbackToken<\/span>(<\/span>context)<\/span>;<\/span>
\n if<\/span> (<\/span>fallbackToken !=<\/span> null<\/span>)<\/span> {<\/span>
\n subscriber.<\/span>onNext<\/span>(<\/span>fallbackToken)<\/span>;<\/span>
\n }<\/span>
\n subscriber.<\/span>onComplete<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span> catch<\/span> (<\/span>Exception<\/span> e)<\/span> {<\/span>
\n subscriber.<\/span>onError<\/span>(<\/span>e)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> restartStream<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber,<\/span>
\n StreamContext<\/span> context)<\/span> {<\/span>
\n \/\/ \u6e05\u7406\u5f53\u524d\u8ba2\u9605<\/span>
\n cleanupSubscriber<\/span>(<\/span>subscriber)<\/span>;<\/span><\/p>\n

\/\/ \u521b\u5efa\u65b0\u7684\u8ba2\u9605<\/span>
\n StreamContext<\/span> newContext =<\/span> new<\/span> StreamContext<\/span>(<\/span>
\n context.<\/span>getStreamId<\/span>(<\/span>)<\/span> +<\/span> "-restarted"<\/span>)<\/span>;<\/span>
\n contexts.<\/span>put<\/span>(<\/span>subscriber,<\/span> newContext)<\/span>;<\/span><\/p>\n

\/\/ \u91cd\u65b0\u542f\u52a8\u6d41\u5904\u7406<\/span>
\n executor.<\/span>submit<\/span>(<\/span>(<\/span>)<\/span> -><\/span> processStream<\/span>(<\/span>subscriber,<\/span> newContext)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

private<\/span> long<\/span> extractDelay<\/span>(<\/span>HandlingResult<\/span> result)<\/span> {<\/span>
\n \/\/ \u4ece\u6d88\u606f\u4e2d\u63d0\u53d6\u5ef6\u8fdf\u65f6\u95f4（\u7b80\u5316\u5b9e\u73b0）<\/span>
\n String<\/span> message =<\/span> result.<\/span>getMessage<\/span>(<\/span>)<\/span>;<\/span>
\n \/\/ \u5b9e\u9645\u5b9e\u73b0\u9700\u8981\u66f4\u590d\u6742\u7684\u89e3\u6790\u903b\u8f91<\/span>
\n return<\/span> 1000L<\/span>;<\/span> \/\/ \u9ed8\u8ba41\u79d2<\/span>
\n }<\/span><\/p>\n

private<\/span> SafeSubscription<\/span> getSubscription<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber)<\/span> {<\/span>
\n return<\/span> (<\/span>SafeSubscription<\/span>)<\/span> subscriber;<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> cleanupSubscriber<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber)<\/span> {<\/span>
\n subscribers.<\/span>remove<\/span>(<\/span>subscriber)<\/span>;<\/span>
\n contexts.<\/span>remove<\/span>(<\/span>subscriber)<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> close<\/span>(<\/span>)<\/span> {<\/span>
\n isClosed =<\/span> true<\/span>;<\/span><\/p>\n

\/\/ \u901a\u77e5\u6240\u6709\u8ba2\u9605\u8005<\/span>
\n for<\/span> (<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber :<\/span> subscribers)<\/span> {<\/span>
\n subscriber.<\/span>onError<\/span>(<\/span>new<\/span> StreamInterruptedException<\/span>(<\/span>
\n StreamInterruptedException.<\/span>InterruptionSource<\/span>.<\/span>SYSTEM_SHUTDOWN,<\/span>
\n false<\/span>,<\/span>
\n contexts.<\/span>get<\/span>(<\/span>subscriber)<\/span>.<\/span>getStreamId<\/span>(<\/span>)<\/span>)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

\/\/ \u6e05\u7406\u8d44\u6e90<\/span>
\n subscribers.<\/span>clear<\/span>(<\/span>)<\/span>;<\/span>
\n contexts.<\/span>clear<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u7b49\u5f85\u6d3b\u8dc3\u6d41\u7ed3\u675f<\/span>
\n awaitTermination<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u5173\u95ed\u6267\u884c\u5668<\/span>
\n executor.<\/span>shutdown<\/span>(<\/span>)<\/span>;<\/span>
\n try<\/span> {<\/span>
\n if<\/span> (<\/span>!<\/span>executor.<\/span>awaitTermination<\/span>(<\/span>10<\/span>,<\/span> TimeUnit<\/span>.<\/span>SECONDS)<\/span>)<\/span> {<\/span>
\n executor.<\/span>shutdownNow<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span> catch<\/span> (<\/span>InterruptedException<\/span> e)<\/span> {<\/span>
\n executor.<\/span>shutdownNow<\/span>(<\/span>)<\/span>;<\/span>
\n Thread<\/span>.<\/span>currentThread<\/span>(<\/span>)<\/span>.<\/span>interrupt<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> awaitTermination<\/span>(<\/span>)<\/span> {<\/span>
\n int<\/span> maxWaitTime =<\/span> 30<\/span>;<\/span> \/\/ \u6700\u591a\u7b49\u5f8530\u79d2<\/span>
\n for<\/span> (<\/span>int<\/span> i =<\/span> 0<\/span>;<\/span> i <<\/span> maxWaitTime &&<\/span> activeStreams.<\/span>get<\/span>(<\/span>)<\/span> ><\/span> 0<\/span>;<\/span> i++<\/span>)<\/span> {<\/span>
\n try<\/span> {<\/span>
\n Thread<\/span>.<\/span>sleep<\/span>(<\/span>1000<\/span>)<\/span>;<\/span>
\n }<\/span> catch<\/span> (<\/span>InterruptedException<\/span> e)<\/span> {<\/span>
\n Thread<\/span>.<\/span>currentThread<\/span>(<\/span>)<\/span>.<\/span>interrupt<\/span>(<\/span>)<\/span>;<\/span>
\n break<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

private<\/span> String<\/span> generateStreamId<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> "stream-"<\/span> +<\/span> UUID.<\/span>randomUUID<\/span>(<\/span>)<\/span>.<\/span>toString<\/span>(<\/span>)<\/span>.<\/span>substring<\/span>(<\/span>0<\/span>,<\/span> 8<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

\/\/ \u8f85\u52a9\u7c7b<\/span>
\n private<\/span> static<\/span> class<\/span> GenericStreamException<\/span> extends<\/span> StreamException<\/span> {<\/span>
\n public<\/span> GenericStreamException<\/span>(<\/span>String<\/span> message,<\/span> Throwable<\/span> cause,<\/span>
\n String<\/span> streamId)<\/span> {<\/span>
\n super<\/span>(<\/span>message,<\/span> ErrorCategory<\/span>.<\/span>CLIENT,<\/span> streamId)<\/span>;<\/span>
\n initCause<\/span>(<\/span>cause)<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> boolean<\/span> isRecoverable<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> false<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> RecoveryStrategy<\/span> getRecoveryStrategy<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> RecoveryStrategy<\/span>.<\/span>FAIL_FAST;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

private<\/span> static<\/span> class<\/span> CircuitBreakerOpenException<\/span> extends<\/span> StreamException<\/span> {<\/span>
\n public<\/span> CircuitBreakerOpenException<\/span>(<\/span>String<\/span> message,<\/span> String<\/span> streamId)<\/span> {<\/span>
\n super<\/span>(<\/span>message,<\/span> ErrorCategory<\/span>.<\/span>CLIENT,<\/span> streamId)<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> boolean<\/span> isRecoverable<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> true<\/span>;<\/span> \/\/ \u65ad\u8def\u5668\u6700\u7ec8\u4f1a\u5173\u95ed<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> RecoveryStrategy<\/span> getRecoveryStrategy<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> RecoveryStrategy<\/span>.<\/span>RETRY_AFTER_DELAY;<\/span>
\n }<\/span>
\n }<\/span>
\n}<\/span><\/p>\n

3.2 \u9ad8\u7ea7\u7279\u6027：\u65ad\u70b9\u7eed\u4f20\u4e0e\u72b6\u6001\u6062\u590d<\/h4>\n

\/**
\n * \u652f\u6301\u65ad\u70b9\u7eed\u4f20\u7684 TokenStream
\n *\/<\/span>
\npublic<\/span> class<\/span> ResumableTokenStream<\/span><<\/span>T<\/span>><\/span><\/span> extends<\/span> SafeTokenStream<\/span><<\/span>T<\/span>><\/span><\/span> {<\/span><\/p>\n

private<\/span> final<\/span> StreamStateStorage<\/span> stateStorage;<\/span>
\n private<\/span> final<\/span> CheckpointStrategy<\/span> checkpointStrategy;<\/span><\/p>\n

public<\/span> ResumableTokenStream<\/span>(<\/span>TokenSource<\/span><<\/span>T<\/span>><\/span><\/span> tokenSource,<\/span>
\n ExecutorService<\/span> executor,<\/span>
\n ExceptionHandlerRegistry<\/span> exceptionHandler,<\/span>
\n StreamRecoveryManager<\/span> recoveryManager,<\/span>
\n StreamStateStorage<\/span> stateStorage,<\/span>
\n CheckpointStrategy<\/span> checkpointStrategy)<\/span> {<\/span>
\n super<\/span>(<\/span>tokenSource,<\/span> executor,<\/span> exceptionHandler,<\/span> recoveryManager)<\/span>;<\/span>
\n this<\/span>.<\/span>stateStorage =<\/span> stateStorage;<\/span>
\n this<\/span>.<\/span>checkpointStrategy =<\/span> checkpointStrategy;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> subscribe<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber)<\/span> {<\/span>
\n \/\/ \u68c0\u67e5\u662f\u5426\u6709\u4fdd\u5b58\u7684\u72b6\u6001<\/span>
\n StreamState<\/span> savedState =<\/span> stateStorage.<\/span>loadState<\/span>(<\/span>
\n getStreamIdFromSubscriber<\/span>(<\/span>subscriber)<\/span>)<\/span>;<\/span><\/p>\n

if<\/span> (<\/span>savedState !=<\/span> null<\/span> &&<\/span> savedState.<\/span>canResume<\/span>(<\/span>)<\/span>)<\/span> {<\/span>
\n \/\/ \u6062\u590d\u8ba2\u9605<\/span>
\n resumeFromState<\/span>(<\/span>subscriber,<\/span> savedState)<\/span>;<\/span>
\n }<\/span> else<\/span> {<\/span>
\n \/\/ \u65b0\u7684\u8ba2\u9605<\/span>
\n super<\/span>.<\/span>subscribe<\/span>(<\/span>subscriber)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> resumeFromState<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber,<\/span>
\n StreamState<\/span> state)<\/span> {<\/span><\/p>\n

StreamContext<\/span> context =<\/span> createContextFromState<\/span>(<\/span>state)<\/span>;<\/span><\/p>\n

\/\/ \u6062\u590dtoken\u6e90\u5230\u68c0\u67e5\u70b9\u4f4d\u7f6e<\/span>
\n getTokenSource<\/span>(<\/span>)<\/span>.<\/span>restoreFromCheckpoint<\/span>(<\/span>state.<\/span>getCheckpoint<\/span>(<\/span>)<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u91cd\u65b0\u8ba2\u9605<\/span>
\n subscribers.<\/span>add<\/span>(<\/span>subscriber)<\/span>;<\/span>
\n contexts.<\/span>put<\/span>(<\/span>subscriber,<\/span> context)<\/span>;<\/span><\/p>\n

subscriber.<\/span>onSubscribe<\/span>(<\/span>new<\/span> ResumableSubscription<\/span>(<\/span>
\n subscriber,<\/span> context,<\/span> state.<\/span>getLastProcessedToken<\/span>(<\/span>)<\/span>)<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u7ee7\u7eed\u5904\u7406<\/span>
\n executor.<\/span>submit<\/span>(<\/span>(<\/span>)<\/span> -><\/span> processResumedStream<\/span>(<\/span>
\n subscriber,<\/span> context,<\/span> state)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

private<\/span> class<\/span> ResumableSubscription<\/span> extends<\/span> SafeSubscription<\/span> {<\/span>
\n private<\/span> final<\/span> T<\/span> lastProcessedToken;<\/span>
\n private<\/span> boolean<\/span> hasDeliveredResumeToken =<\/span> false<\/span>;<\/span><\/p>\n

public<\/span> ResumableSubscription<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber,<\/span>
\n StreamContext<\/span> context,<\/span>
\n T<\/span> lastProcessedToken)<\/span> {<\/span>
\n super<\/span>(<\/span>subscriber,<\/span> context)<\/span>;<\/span>
\n this<\/span>.<\/span>lastProcessedToken =<\/span> lastProcessedToken;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> request<\/span>(<\/span>long<\/span> n)<\/span> {<\/span>
\n if<\/span> (<\/span>!<\/span>hasDeliveredResumeToken &&<\/span> lastProcessedToken !=<\/span> null<\/span>)<\/span> {<\/span>
\n \/\/ \u9996\u5148\u53d1\u9001\u6062\u590d\u6807\u8bb0<\/span>
\n subscriber.<\/span>onNext<\/span>(<\/span>createResumeToken<\/span>(<\/span>)<\/span>)<\/span>;<\/span>
\n hasDeliveredResumeToken =<\/span> true<\/span>;<\/span>
\n n—<\/span>;<\/span> \/\/ \u51cf\u5c11\u4e00\u4e2a\u8bf7\u6c42\u8ba1\u6570<\/span>
\n }<\/span><\/p>\n

if<\/span> (<\/span>n ><\/span> 0<\/span>)<\/span> {<\/span>
\n super<\/span>.<\/span>request<\/span>(<\/span>n)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

private<\/span> T<\/span> createResumeToken<\/span>(<\/span>)<\/span> {<\/span>
\n \/\/ \u521b\u5efa\u7279\u6b8a\u7684\u6062\u590dtoken<\/span>
\n return<\/span> (<\/span>T<\/span>)<\/span> new<\/span> ResumeToken<\/span><<\/span>><\/span><\/span>(<\/span>lastProcessedToken)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> processResumedStream<\/span>(<\/span>Subscriber<\/span><<\/span>?<\/span> super<\/span> T<\/span>><\/span><\/span> subscriber,<\/span>
\n StreamContext<\/span> context,<\/span>
\n StreamState<\/span> state)<\/span> {<\/span><\/p>\n

\/\/ \u8df3\u8fc7\u5df2\u7ecf\u5904\u7406\u8fc7\u7684token<\/span>
\n skipProcessedTokens<\/span>(<\/span>state.<\/span>getProcessedCount<\/span>(<\/span>)<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u7ee7\u7eed\u6b63\u5e38\u5904\u7406<\/span>
\n super<\/span>.<\/span>processStream<\/span>(<\/span>subscriber,<\/span> context)<\/span>;<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> skipProcessedTokens<\/span>(<\/span>int<\/span> count)<\/span> {<\/span>
\n for<\/span> (<\/span>int<\/span> i =<\/span> 0<\/span>;<\/span> i <<\/span> count;<\/span> i++<\/span>)<\/span> {<\/span>
\n try<\/span> {<\/span>
\n getTokenSource<\/span>(<\/span>)<\/span>.<\/span>nextToken<\/span>(<\/span>)<\/span>;<\/span> \/\/ \u8df3\u8fc7<\/span>
\n }<\/span> catch<\/span> (<\/span>Exception<\/span> e)<\/span> {<\/span>
\n \/\/ \u8bb0\u5f55\u4f46\u7ee7\u7eed<\/span>
\n logger.<\/span>warn<\/span>(<\/span>"Failed to skip token {}"<\/span>,<\/span> i,<\/span> e)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u5b9a\u671f\u4fdd\u5b58\u72b6\u6001<\/span>
\n private<\/span> void<\/span> createCheckpoint<\/span>(<\/span>StreamContext<\/span> context,<\/span>
\n T<\/span> lastToken,<\/span>
\n int<\/span> processedCount)<\/span> {<\/span><\/p>\n

if<\/span> (<\/span>checkpointStrategy.<\/span>shouldCheckpoint<\/span>(<\/span>context,<\/span> lastToken)<\/span>)<\/span> {<\/span>
\n StreamState<\/span> state =<\/span> new<\/span> StreamState<\/span>(<\/span>
\n context.<\/span>getStreamId<\/span>(<\/span>)<\/span>,<\/span>
\n getTokenSource<\/span>(<\/span>)<\/span>.<\/span>createCheckpoint<\/span>(<\/span>)<\/span>,<\/span>
\n lastToken,<\/span>
\n processedCount,<\/span>
\n Instant<\/span>.<\/span>now<\/span>(<\/span>)<\/span>
\n )<\/span>;<\/span><\/p>\n

stateStorage.<\/span>saveState<\/span>(<\/span>state)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u6d41\u72b6\u6001\u7c7b<\/span>
\n public<\/span> static<\/span> class<\/span> StreamState<\/span> {<\/span>
\n private<\/span> final<\/span> String<\/span> streamId;<\/span>
\n private<\/span> final<\/span> Checkpoint<\/span> checkpoint;<\/span>
\n private<\/span> final<\/span> Object<\/span> lastProcessedToken;<\/span>
\n private<\/span> final<\/span> int<\/span> processedCount;<\/span>
\n private<\/span> final<\/span> Instant<\/span> checkpointTime;<\/span><\/p>\n

public<\/span> StreamState<\/span>(<\/span>String<\/span> streamId,<\/span> Checkpoint<\/span> checkpoint,<\/span>
\n Object<\/span> lastProcessedToken,<\/span> int<\/span> processedCount,<\/span>
\n Instant<\/span> checkpointTime)<\/span> {<\/span>
\n this<\/span>.<\/span>streamId =<\/span> streamId;<\/span>
\n this<\/span>.<\/span>checkpoint =<\/span> checkpoint;<\/span>
\n this<\/span>.<\/span>lastProcessedToken =<\/span> lastProcessedToken;<\/span>
\n this<\/span>.<\/span>processedCount =<\/span> processedCount;<\/span>
\n this<\/span>.<\/span>checkpointTime =<\/span> checkpointTime;<\/span>
\n }<\/span><\/p>\n

public<\/span> boolean<\/span> canResume<\/span>(<\/span>)<\/span> {<\/span>
\n return<\/span> checkpoint !=<\/span> null<\/span> &&<\/span>
\n checkpointTime.<\/span>isAfter<\/span>(<\/span>Instant<\/span>.<\/span>now<\/span>(<\/span>)<\/span>.<\/span>minus<\/span>(<\/span>Duration<\/span>.<\/span>ofHours<\/span>(<\/span>1<\/span>)<\/span>)<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u6062\u590dtoken\u6807\u8bb0<\/span>
\n public<\/span> static<\/span> class<\/span> ResumeToken<\/span><<\/span>T<\/span>><\/span><\/span> {<\/span>
\n private<\/span> final<\/span> T<\/span> previousToken;<\/span>
\n private<\/span> final<\/span> Instant<\/span> resumeTime;<\/span><\/p>\n

public<\/span> ResumeToken<\/span>(<\/span>T<\/span> previousToken)<\/span> {<\/span>
\n this<\/span>.<\/span>previousToken =<\/span> previousToken;<\/span>
\n this<\/span>.<\/span>resumeTime =<\/span> Instant<\/span>.<\/span>now<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n}<\/span><\/p>\n

\u56db\u3001\u6700\u4f73\u5b9e\u8df5\u4e0e\u6a21\u5f0f\u603b\u7ed3<\/h3>\n

4.1 \u5f02\u5e38\u5904\u7406\u6700\u4f73\u5b9e\u8df5<\/h4>\n

\/**
\n * TokenStream \u5f02\u5e38\u5904\u7406\u7684\u6700\u4f73\u5b9e\u8df5
\n *\/<\/span>
\npublic<\/span> class<\/span> TokenStreamBestPractices<\/span> {<\/span><\/p>\n

\/\/ \u5b9e\u8df51：\u5206\u5c42\u5f02\u5e38\u5904\u7406<\/span>
\n public<\/span> static<\/span> class<\/span> LayeredExceptionHandling<\/span> {<\/span><\/p>\n

public<\/span> void<\/span> processStreamWithLayers<\/span>(<\/span>)<\/span> {<\/span>
\n try<\/span> {<\/span>
\n \/\/ \u5916\u5c42：\u8fde\u63a5\u548c\u521d\u59cb\u5316\u5f02\u5e38<\/span>
\n TokenStream<\/span><<\/span>String<\/span>><\/span><\/span> stream =<\/span> createTokenStream<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

stream.<\/span>subscribe<\/span>(<\/span>new<\/span> Subscriber<\/span><<\/span>String<\/span>><\/span><\/span>(<\/span>)<\/span> {<\/span>
\n @Override<\/span>
\n public<\/span> void<\/span> onSubscribe<\/span>(<\/span>Subscription<\/span> s)<\/span> {<\/span>
\n \/\/ \u4e2d\u5c42：\u8ba2\u9605\u548c\u8bf7\u6c42\u5f02\u5e38<\/span>
\n try<\/span> {<\/span>
\n s.<\/span>request<\/span>(<\/span>1<\/span>)<\/span>;<\/span>
\n }<\/span> catch<\/span> (<\/span>IllegalArgumentException<\/span> e)<\/span> {<\/span>
\n handleRequestError<\/span>(<\/span>e)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> onNext<\/span>(<\/span>String<\/span> token)<\/span> {<\/span>
\n \/\/ \u5185\u5c42：\u6570\u636e\u5904\u7406\u5f02\u5e38<\/span>
\n try<\/span> {<\/span>
\n processToken<\/span>(<\/span>token)<\/span>;<\/span>
\n }<\/span> catch<\/span> (<\/span>ProcessingException<\/span> e)<\/span> {<\/span>
\n handleTokenError<\/span>(<\/span>e,<\/span> token)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> onError<\/span>(<\/span>Throwable<\/span> t)<\/span> {<\/span>
\n \/\/ \u9519\u8bef\u5904\u7406\u5c42<\/span>
\n handleStreamError<\/span>(<\/span>t)<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> onComplete<\/span>(<\/span>)<\/span> {<\/span>
\n handleCompletion<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>)<\/span>;<\/span><\/p>\n

}<\/span> catch<\/span> (<\/span>StreamCreationException<\/span> e)<\/span> {<\/span>
\n handleCreationError<\/span>(<\/span>e)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u5b9e\u8df52：\u4f18\u96c5\u964d\u7ea7<\/span>
\n public<\/span> static<\/span> class<\/span> GracefulDegradation<\/span> {<\/span><\/p>\n

private<\/span> final<\/span> TokenStream<\/span><<\/span>String<\/span>><\/span><\/span> primaryStream;<\/span>
\n private<\/span> final<\/span> TokenStream<\/span><<\/span>String<\/span>><\/span><\/span> fallbackStream;<\/span>
\n private<\/span> final<\/span> DegradationPolicy<\/span> degradationPolicy;<\/span><\/p>\n

public<\/span> void<\/span> startStreamWithFallback<\/span>(<\/span>Subscriber<\/span><<\/span>String<\/span>><\/span><\/span> subscriber)<\/span> {<\/span>
\n AtomicBoolean<\/span> fallbackActive =<\/span> new<\/span> AtomicBoolean<\/span>(<\/span>false<\/span>)<\/span>;<\/span><\/p>\n

Subscriber<\/span><<\/span>String<\/span>><\/span><\/span> wrappedSubscriber =<\/span> new<\/span> Subscriber<\/span><<\/span>String<\/span>><\/span><\/span>(<\/span>)<\/span> {<\/span>
\n @Override<\/span>
\n public<\/span> void<\/span> onSubscribe<\/span>(<\/span>Subscription<\/span> s)<\/span> {<\/span>
\n subscriber.<\/span>onSubscribe<\/span>(<\/span>s)<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> onNext<\/span>(<\/span>String<\/span> token)<\/span> {<\/span>
\n subscriber.<\/span>onNext<\/span>(<\/span>token)<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> onError<\/span>(<\/span>Throwable<\/span> t)<\/span> {<\/span>
\n if<\/span> (<\/span>!<\/span>fallbackActive.<\/span>get<\/span>(<\/span>)<\/span> &&<\/span>
\n degradationPolicy.<\/span>shouldUseFallback<\/span>(<\/span>t)<\/span>)<\/span> {<\/span><\/p>\n

fallbackActive.<\/span>set<\/span>(<\/span>true<\/span>)<\/span>;<\/span>
\n logger.<\/span>warn<\/span>(<\/span>"Switching to fallback stream"<\/span>,<\/span> t)<\/span>;<\/span><\/p>\n

\/\/ \u5207\u6362\u5230\u964d\u7ea7\u6d41<\/span>
\n fallbackStream.<\/span>subscribe<\/span>(<\/span>this<\/span>)<\/span>;<\/span>
\n }<\/span> else<\/span> {<\/span>
\n subscriber.<\/span>onError<\/span>(<\/span>t)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> onComplete<\/span>(<\/span>)<\/span> {<\/span>
\n subscriber.<\/span>onComplete<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>;<\/span><\/p>\n

primaryStream.<\/span>subscribe<\/span>(<\/span>wrappedSubscriber)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u5b9e\u8df53：\u76d1\u63a7\u548c\u6307\u6807<\/span>
\n public<\/span> static<\/span> class<\/span> MonitoringAndMetrics<\/span> {<\/span><\/p>\n

private<\/span> final<\/span> MeterRegistry<\/span> meterRegistry;<\/span>
\n private<\/span> final<\/span> Tracer<\/span> tracer;<\/span><\/p>\n

public<\/span> TokenStream<\/span><<\/span>String<\/span>><\/span><\/span> createMonitoredStream<\/span>(<\/span>
\n TokenStream<\/span><<\/span>String<\/span>><\/span><\/span> delegate)<\/span> {<\/span><\/p>\n

return<\/span> new<\/span> TokenStream<\/span><<\/span>String<\/span>><\/span><\/span>(<\/span>)<\/span> {<\/span>
\n private<\/span> final<\/span> Timer<\/span> streamTimer =<\/span> Timer<\/span>.<\/span>builder<\/span>(<\/span>"token.stream.duration"<\/span>)<\/span>
\n .<\/span>register<\/span>(<\/span>meterRegistry)<\/span>;<\/span><\/p>\n

private<\/span> final<\/span> Counter<\/span> tokenCounter =<\/span> Counter<\/span>.<\/span>builder<\/span>(<\/span>"token.count"<\/span>)<\/span>
\n .<\/span>register<\/span>(<\/span>meterRegistry)<\/span>;<\/span><\/p>\n

private<\/span> final<\/span> Counter<\/span> errorCounter =<\/span> Counter<\/span>.<\/span>builder<\/span>(<\/span>"token.stream.errors"<\/span>)<\/span>
\n .<\/span>tag<\/span>(<\/span>"type"<\/span>,<\/span> "stream"<\/span>)<\/span>
\n .<\/span>register<\/span>(<\/span>meterRegistry)<\/span>;<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> subscribe<\/span>(<\/span>Subscriber<\/span><<\/span>String<\/span>><\/span><\/span> subscriber)<\/span> {<\/span>
\n Timer.<\/span>Sample<\/span> sample =<\/span> Timer<\/span>.<\/span>start<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

delegate.<\/span>subscribe<\/span>(<\/span>new<\/span> Subscriber<\/span><<\/span>String<\/span>><\/span><\/span>(<\/span>)<\/span> {<\/span>
\n private<\/span> Span<\/span> span;<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> onSubscribe<\/span>(<\/span>Subscription<\/span> s)<\/span> {<\/span>
\n \/\/ \u5f00\u59cb\u8ffd\u8e2a<\/span>
\n span =<\/span> tracer.<\/span>buildSpan<\/span>(<\/span>"token.stream"<\/span>)<\/span>
\n .<\/span>startSpan<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

subscriber.<\/span>onSubscribe<\/span>(<\/span>s)<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> onNext<\/span>(<\/span>String<\/span> token)<\/span> {<\/span>
\n tokenCounter.<\/span>increment<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u8bb0\u5f55token\u5927\u5c0f<\/span>
\n DistributionSummary<\/span>.<\/span>builder<\/span>(<\/span>"token.size"<\/span>)<\/span>
\n .<\/span>register<\/span>(<\/span>meterRegistry)<\/span>
\n .<\/span>record<\/span>(<\/span>token.<\/span>length<\/span>(<\/span>)<\/span>)<\/span>;<\/span><\/p>\n

subscriber.<\/span>onNext<\/span>(<\/span>token)<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> onError<\/span>(<\/span>Throwable<\/span> t)<\/span> {<\/span>
\n errorCounter.<\/span>increment<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u8bb0\u5f55\u9519\u8bef\u6807\u7b7e<\/span>
\n span.<\/span>setTag<\/span>(<\/span>"error"<\/span>,<\/span> true<\/span>)<\/span>;<\/span>
\n span.<\/span>log<\/span>(<\/span>Map<\/span>.<\/span>of<\/span>(<\/span>
\n "error.message"<\/span>,<\/span> t.<\/span>getMessage<\/span>(<\/span>)<\/span>,<\/span>
\n "error.type"<\/span>,<\/span> t.<\/span>getClass<\/span>(<\/span>)<\/span>.<\/span>getName<\/span>(<\/span>)<\/span>
\n )<\/span>)<\/span>;<\/span><\/p>\n

span.<\/span>finish<\/span>(<\/span>)<\/span>;<\/span>
\n sample.<\/span>stop<\/span>(<\/span>streamTimer)<\/span>;<\/span><\/p>\n

subscriber.<\/span>onError<\/span>(<\/span>t)<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> onComplete<\/span>(<\/span>)<\/span> {<\/span>
\n span.<\/span>finish<\/span>(<\/span>)<\/span>;<\/span>
\n sample.<\/span>stop<\/span>(<\/span>streamTimer)<\/span>;<\/span><\/p>\n

subscriber.<\/span>onComplete<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

\/\/ \u5b9e\u8df54：\u8d44\u6e90\u6e05\u7406<\/span>
\n public<\/span> static<\/span> class<\/span> ResourceCleanupPattern<\/span> implements<\/span> AutoCloseable<\/span> {<\/span><\/p>\n

private<\/span> final<\/span> List<\/span><<\/span>AutoCloseable<\/span>><\/span><\/span> resources =<\/span> new<\/span> ArrayList<\/span><<\/span>><\/span><\/span>(<\/span>)<\/span>;<\/span>
\n private<\/span> final<\/span> ScheduledExecutorService<\/span> cleanupExecutor;<\/span>
\n private<\/span> final<\/span> AtomicBoolean<\/span> closed =<\/span> new<\/span> AtomicBoolean<\/span>(<\/span>false<\/span>)<\/span>;<\/span><\/p>\n

public<\/span> ResourceCleanupPattern<\/span>(<\/span>)<\/span> {<\/span>
\n this<\/span>.<\/span>cleanupExecutor =<\/span> Executors<\/span>.<\/span>newSingleThreadScheduledExecutor<\/span>(<\/span>
\n r -><\/span> new<\/span> Thread<\/span>(<\/span>r,<\/span> "stream-cleanup"<\/span>)<\/span>
\n )<\/span>;<\/span><\/p>\n

\/\/ \u5b9a\u671f\u68c0\u67e5\u6cc4\u6f0f<\/span>
\n cleanupExecutor.<\/span>scheduleAtFixedRate<\/span>(<\/span>
\n this<\/span>::<\/span>checkForLeaks<\/span>,<\/span>
\n 5<\/span>,<\/span> 60<\/span>,<\/span> TimeUnit<\/span>.<\/span>SECONDS)<\/span>;<\/span><\/p>\n

\/\/ \u6dfb\u52a0\u5173\u95ed\u94a9\u5b50<\/span>
\n Runtime<\/span>.<\/span>getRuntime<\/span>(<\/span>)<\/span>.<\/span>addShutdownHook<\/span>(<\/span>new<\/span> Thread<\/span>(<\/span>this<\/span>::<\/span>safeClose<\/span>)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

public<\/span> <<\/span>T<\/span> extends<\/span> AutoCloseable<\/span>><\/span><\/span> T<\/span> trackResource<\/span>(<\/span>T<\/span> resource)<\/span> {<\/span>
\n resources.<\/span>add<\/span>(<\/span>resource)<\/span>;<\/span>
\n return<\/span> resource;<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> checkForLeaks<\/span>(<\/span>)<\/span> {<\/span>
\n \/\/ \u68c0\u67e5\u672a\u5173\u95ed\u7684\u8d44\u6e90<\/span>
\n \/\/ \u5b9e\u9645\u5b9e\u73b0\u9700\u8981\u66f4\u590d\u6742\u7684\u68c0\u6d4b\u903b\u8f91<\/span>
\n logger.<\/span>debug<\/span>(<\/span>"Active resources: {}"<\/span>,<\/span> resources.<\/span>size<\/span>(<\/span>)<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

@Override<\/span>
\n public<\/span> void<\/span> close<\/span>(<\/span>)<\/span> {<\/span>
\n safeClose<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span><\/p>\n

private<\/span> void<\/span> safeClose<\/span>(<\/span>)<\/span> {<\/span>
\n if<\/span> (<\/span>!<\/span>closed.<\/span>compareAndSet<\/span>(<\/span>false<\/span>,<\/span> true<\/span>)<\/span>)<\/span> {<\/span>
\n return<\/span>;<\/span> \/\/ \u5df2\u7ecf\u5173\u95ed<\/span>
\n }<\/span><\/p>\n

\/\/ \u5173\u95ed\u6e05\u7406\u6267\u884c\u5668<\/span>
\n cleanupExecutor.<\/span>shutdown<\/span>(<\/span>)<\/span>;<\/span><\/p>\n

\/\/ \u9006\u5e8f\u5173\u95ed\u8d44\u6e90（\u4f9d\u8d56\u5173\u7cfb）<\/span>
\n ListIterator<\/span><<\/span>AutoCloseable<\/span>><\/span><\/span> iterator =<\/span>
\n resources.<\/span>listIterator<\/span>(<\/span>resources.<\/span>size<\/span>(<\/span>)<\/span>)<\/span>;<\/span><\/p>\n

while<\/span> (<\/span>iterator.<\/span>hasPrevious<\/span>(<\/span>)<\/span>)<\/span> {<\/span>
\n AutoCloseable<\/span> resource =<\/span> iterator.<\/span>previous<\/span>(<\/span>)<\/span>;<\/span>
\n try<\/span> {<\/span>
\n resource.<\/span>close<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span> catch<\/span> (<\/span>Exception<\/span> e)<\/span> {<\/span>
\n logger.<\/span>warn<\/span>(<\/span>"Failed to close resource"<\/span>,<\/span> e)<\/span>;<\/span>
\n }<\/span>
\n }<\/span><\/p>\n

resources.<\/span>clear<\/span>(<\/span>)<\/span>;<\/span>
\n }<\/span>
\n }<\/span>
\n}<\/span><\/p>\n

4.2 \u9762\u8bd5\u8981\u70b9\u603b\u7ed3<\/h4>\n

Q1: \u4ec0\u4e48\u662f TokenStream？<\/p>\n

A: TokenStream \u662f\u5904\u7406\u5927\u578b\u8bed\u8a00\u6a21\u578b\u6d41\u5f0f\u54cd\u5e94\u7684\u6838\u5fc3\u62bd\u8c61，\u5177\u6709\u4ee5\u4e0b\u7279\u70b9：<\/p>\n

  • \u5f02\u6b65\u6d41\u5f0f\u5904\u7406：\u652f\u6301\u975e\u963b\u585e\u7684\u589e\u91cf\u6570\u636e\u751f\u6210<\/li>\n
  • Token \u5e8f\u5217\u5316：\u5c06\u6587\u672c\u5206\u89e3\u4e3a\u8bed\u4e49\u5355\u5143\u8fdb\u884c\u4f20\u8f93<\/li>\n
  • \u80cc\u538b\u652f\u6301：\u57fa\u4e8e\u54cd\u5e94\u5f0f\u6d41\u89c4\u8303，\u652f\u6301\u6d41\u91cf\u63a7\u5236<\/li>\n
  • \u9519\u8bef\u4f20\u64ad：\u5728\u6d41\u4e2d\u4f20\u9012\u5f02\u5e38\u800c\u4e0d\u4e2d\u65ad\u6574\u4e2a\u7cfb\u7edf<\/li>\n
  • \u53ef\u6062\u590d\u6027：\u652f\u6301\u4ece\u6545\u969c\u70b9\u6062\u590d\u5904\u7406<\/li>\n

    Q2: \u5982\u4f55\u5904\u7406\u6d41\u5f0f\u54cd\u5e94\u4e2d\u7684\u5f02\u5e38？<\/p>\n

    A: \u91c7\u7528\u5206\u5c42\u5f02\u5e38\u5904\u7406\u7b56\u7565：<\/p>\n

  • \n

    \u5f02\u5e38\u5206\u7c7b：<\/p>\n

    \/\/ \u7f51\u7edc\u5f02\u5e38：\u53ef\u6062\u590d，\u9700\u8981\u91cd\u8bd5\u7b56\u7565<\/span>
    \n\/\/ \u4e1a\u52a1\u5f02\u5e38：\u53ef\u80fd\u9700\u8981\u964d\u7ea7\u5904\u7406<\/span>
    \n\/\/ \u4e2d\u65ad\u5f02\u5e38：\u9700\u8981\u6e05\u7406\u8d44\u6e90\u5e76\u901a\u77e5\u7528\u6237<\/span><\/p>\n<\/li>\n

  • \n

    \u5904\u7406\u7b56\u7565：<\/p>\n