當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

解耦,未解耦的区别_幂等与时间解耦之旅

發布時間：2023/12/3 编程问答 28 豆豆

生活随笔收集整理的這篇文章主要介紹了解耦,未解耦的区别_幂等与时间解耦之旅小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

解耦,未解耦的區別

HTTP中的冪等性意味著相同的請求可以執行多次，效果與僅執行一次一樣。如果用新資源替換某個資源的當前狀態，則無論您執行多少次，最終狀態都將與您僅執行一次相同。舉一個更具體的例子：刪除用戶是冪等的，因為無論您通過唯一標識符刪除給定用戶多少次，最終該用戶都會被刪除。另一方面，創建新用戶不是冪等的，因為兩次請求該操作將創建兩個用戶。用HTTP術語來說是RFC 2616：9.1.2等冪方法必須說的：

9.1.2等冪方法

方法還可以具有“ 冪等 ”的特性，因為[…] N> 0個相同請求的副作用與單個請求的副作用相同。 GET，HEAD，PUT和DELETE方法共享此屬性。同樣，方法OPTIONS和TRACE不應有副作用，因此本質上是冪等的。

時間耦合是系統的不良特性，其中正確的行為隱含地取決于時間維度。用簡單的英語來說，這可能意味著例如系統僅在所有組件同時存在時才起作用。阻塞請求-響應通信（ReST，SOAP或任何其他形式的RPC）要求客戶端和服務器同時可用，這就是這種效果的一個示例。

基本了解這些概念的含義后，我們來看一個簡單的案例研究- 大型多人在線角色扮演游戲。我們的人工用例如下：玩家發送優質短信，以在游戲內購買虛擬劍。交付SMS時將調用我們的HTTP網關，我們需要通知部署在另一臺計算機上的InventoryService 。當前的API涉及ReST，其外觀如下：

@Slf4j @RestController class SmsController {private final RestOperations restOperations;@Autowiredpublic SmsController(RestOperations restOperations) {this.restOperations = restOperations;}@RequestMapping(value = "/sms/{phoneNumber}", method = POST)public void handleSms(@PathVariable String phoneNumber) {Optional<Player> maybePlayer = phoneNumberToPlayer(phoneNumber);maybePlayer.map(Player::getId).map(this::purchaseSword).orElseThrow(() -> new IllegalArgumentException("Unknown player for phone number " + phoneNumber));}private long purchaseSword(long playerId) {Sword sword = new Sword();HttpEntity<String> entity = new HttpEntity<>(sword.toJson(), jsonHeaders());restOperations.postForObject("http://inventory:8080/player/{playerId}/inventory",entity, Object.class, playerId);return playerId;}private HttpHeaders jsonHeaders() {HttpHeaders headers = new HttpHeaders();headers.setContentType(MediaType.APPLICATION_JSON);return headers;}private Optional<Player> phoneNumberToPlayer(String phoneNumber) {//...} }

依次產生類似于以下內容的請求：

> POST /player/123123/inventory HTTP/1.1 > Host: inventory:8080 > Content-type: application/json > > {"type": "sword", "strength": 100, ...}< HTTP/1.1 201 Created < Content-Length: 75 < Content-Type: application/json;charset=UTF-8 < Location: http://inventory:8080/player/123123/inventory/1

這很簡單。 SmsController只需通過發布購買的劍SmsController適當的數據轉發到SmsController inventory:8080服務。該服務立即或201 Created返回201 Created HTTP響應，確認操作成功。此外，還會創建并返回到資源的鏈接，因此您可以對其進行查詢。有人會說：ReST是最新技術。但是，如果您至少關心客戶的錢并了解什么是ACID（比特幣交易所還必須學習的東西：請參閱[1] ， [2] ， [3]和[4] ）–該API也是易碎，容易出錯。想象所有這些類型的錯誤：

您的請求從未到達inventory服務器

您的請求已到達服務器，但被拒絕

服務器接受連接，但無法讀取請求

服務器讀取請求但掛起

服務器處理了請求，但發送響應失敗

服務器發送了200 OK響應，但丟失了，您再也沒有收到

收到服務器的響應，但客戶端無法處理它

服務器的響應已發送，但客戶端更早超時

在所有這些情況下，您僅在客戶端獲得一個異常，而您不知道服務器的狀態是什么。從技術上講，您應該重試失敗的請求，但是由于POST不具有冪等性，因此您最終可能會用一把以上的劍來獎勵玩家（在5-8情況下）。但是，如果不重試，您可能會失去游戲玩家的金錢而又不給他他寶貴的神器。肯定有更好的辦法。

將POST轉換為冪等PUT

在某些情況下，通過將ID生成基本上從服務器轉移到客戶端，從POST轉換為冪等PUT會非常簡單。使用POST的是服務器生成劍的ID，并將其發送到Location標頭中的客戶端。事實證明，在客戶端急切地生成UUID并稍稍更改語義加上在服務器端強制執行一些約束就足夠了：

private long purchaseSword(long playerId) {Sword sword = new Sword();UUID uuid = sword.getUuid();HttpEntity<String> entity = new HttpEntity<>(sword.toJson(), jsonHeaders());asyncRetryExecutor.withMaxRetries(10).withExponentialBackoff(100, 2.0).doWithRetry(ctx ->restOperations.put("http://inventory:8080/player/{playerId}/inventory/{uuid}",entity, playerId, uuid));return playerId; }

該API如下所示：

> PUT /player/123123/inventory/45e74f80-b2fb-11e4-ab27-0800200c9a66 HTTP/1.1 > Host: inventory:8080 > Content-type: application/json;charset=UTF-8 > > {"type": "sword", "strength": 100, ...}< HTTP/1.1 201 Created < Content-Length: 75 < Content-Type: application/json;charset=UTF-8 < Location: http://inventory:8080/player/123123/inventory/45e74f80-b2fb-11e4-ab27-0800200c9a66

為什么這么大？簡單地說（不需要雙關語），客戶端現在可以根據需要重試PUT請求多次。服務器首次收到PUT時，會將劍以客戶端生成的UUID（ 45e74f80-b2fb-11e4-ab27-0800200c9a66 ）作為主鍵45e74f80-b2fb-11e4-ab27-0800200c9a66在數據庫中。在第二次嘗試PUT的情況下，我們可以更新或拒絕該請求。使用POST不可能，因為每個請求都被視為購買新劍–現在我們可以跟蹤是否已經有這樣的PUT。我們只需要記住，后續的PUT并不是錯誤，而是更新請求：

@RestController @Slf4j public class InventoryController {private final PlayerRepository playerRepository;@Autowiredpublic InventoryController(PlayerRepository playerRepository) {this.playerRepository = playerRepository;}@RequestMapping(value = "/player/{playerId}/inventory/{invId}", method = PUT)@Transactionalpublic void addSword(@PathVariable UUID playerId, @PathVariable UUID invId) {playerRepository.findOne(playerId).addSwordWithId(invId);}}interface PlayerRepository extends JpaRepository<Player, UUID> {}@lombok.Data @lombok.AllArgsConstructor @lombok.NoArgsConstructor @Entity class Sword {@Id@Convert(converter = UuidConverter.class)UUID id;int strength;@Overridepublic boolean equals(Object o) {if (this == o) return true;if (!(o instanceof Sword)) return false;Sword sword = (Sword) o;return id.equals(sword.id);}@Overridepublic int hashCode() {return id.hashCode();} }@Data @Entity class Player {@Id@Convert(converter = UuidConverter.class)UUID id = UUID.randomUUID();@OneToMany(cascade = ALL, fetch = EAGER)@JoinColumn(name="player_id")Set<Sword> swords = new HashSet<>();public Player addSwordWithId(UUID id) {swords.add(new Sword(id, 100));return this;}}

上面的代碼片段中很少有快捷方式，例如直接將存儲庫注入到控制器，以及使用@Transactional注釋。但是你明白了。還要注意，假設沒有完全同時插入兩個具有相同UUID的劍，此代碼相當樂觀。否則將發生約束違例異常。

旁注1：我在控制器和JPA模型中都使用UUID類型。開箱即用不支持它們，對于JPA，您需要自定義轉換器：

public class UuidConverter implements AttributeConverter<UUID, String> {@Overridepublic String convertToDatabaseColumn(UUID attribute) {return attribute.toString();}@Overridepublic UUID convertToEntityAttribute(String dbData) {return UUID.fromString(dbData);} }

對于Spring MVC同樣（僅單向）：

@Bean GenericConverter uuidConverter() {return new GenericConverter() {@Overridepublic Set<ConvertiblePair> getConvertibleTypes() {return Collections.singleton(new ConvertiblePair(String.class, UUID.class));}@Overridepublic Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {return UUID.fromString(source.toString());}}; }

附注2：如果無法更改客戶端，則可以通過將每個請求的哈希存儲在服務器端來跟蹤重復項。這樣，當多次發送同一請求（客戶端重試）時，它將被忽略。但是有時我們可能會有合法的用例，可以兩次發送完全相同的請求（例如，在短時間內購買兩把劍）。

時間耦合–客戶不可用

您認為自己很聰明，但是僅重試就不夠了。首先，客戶端可以在重新嘗試失敗的請求時死亡。如果服務器嚴重損壞或關閉，重試可能要花費幾分鐘甚至幾小時。您不能僅僅因為下游依賴項之一關閉而就阻止了傳入的HTTP請求-如果可能，您必須在后臺異步處理此類請求。但是，延長重試時間會增加客戶端死亡或重新啟動的可能性，這可能會使我們的請求松動。想象一下，我們收到了優質的SMS，但是InventoryService目前處于關閉狀態。我們可以在第二，第二，第四等之后重試，但是如果InventoryService停機了幾個小時又碰巧我們的服務也重新啟動了怎么辦？我們只是失去了短信和劍從未被賦予玩家的機會。

解決此問題的方法是先保留未決請求，然后在后臺處理它。收到SMS消息后，我們幾乎沒有將玩家ID存儲在名為“ pending_purchases數據庫表中。后臺調度程序或事件喚醒異步線程，該線程將收集所有未完成的購買并將嘗試將其發送到InventoryService （甚至可能以批處理方式？）每隔一分鐘甚至一秒鐘運行一次的周期性批處理線程，并收集所有未完成的請求將不可避免地導致延遲和不必要數據庫流量。因此，我打算使用Quartz調度程序，它將為每個待處理的請求調度重試作業：

@Slf4j @RestController class SmsController {private Scheduler scheduler;@Autowiredpublic SmsController(Scheduler scheduler) {this.scheduler = scheduler;}@RequestMapping(value = "/sms/{phoneNumber}", method = POST)public void handleSms(@PathVariable String phoneNumber) {phoneNumberToPlayer(phoneNumber).map(Player::getId).map(this::purchaseSword).orElseThrow(() -> new IllegalArgumentException("Unknown player for phone number " + phoneNumber));}private UUID purchaseSword(UUID playerId) {UUID swordId = UUID.randomUUID();InventoryAddJob.scheduleOn(scheduler, Duration.ZERO, playerId, swordId);return swordId;}//...}

和工作本身：

@Slf4j public class InventoryAddJob implements Job {@Autowired private RestOperations restOperations;@lombok.Setter private UUID invId;@lombok.Setter private UUID playerId;@Overridepublic void execute(JobExecutionContext context) throws JobExecutionException {try {tryPurchase();} catch (Exception e) {Duration delay = Duration.ofSeconds(5);log.error("Can't add to inventory, will retry in {}", delay, e);scheduleOn(context.getScheduler(), delay, playerId, invId);}}private void tryPurchase() {restOperations.put(/*...*/);}public static void scheduleOn(Scheduler scheduler, Duration delay, UUID playerId, UUID invId) {try {JobDetail job = newJob().ofType(InventoryAddJob.class).usingJobData("playerId", playerId.toString()).usingJobData("invId", invId.toString()).build();Date runTimestamp = Date.from(Instant.now().plus(delay));Trigger trigger = newTrigger().startAt(runTimestamp).build();scheduler.scheduleJob(job, trigger);} catch (SchedulerException e) {throw new RuntimeException(e);}}}

每當我們收到優質的SMS時，我們都會安排異步作業立即執行。 Quartz將負責持久性（如果應用程序關閉，則在重新啟動后將盡快執行作業）。而且，如果該特定實例出現故障，則另一個可以承擔這項工作–或我們可以形成集群并在它們之間進行負載平衡請求：一個實例接收SMS，另一個實例在InventoryService請求劍。顯然，如果HTTP調用失敗，則稍后重新安排重試時間，一切都是事務性的且具有故障保護功能。在實際代碼中，您可能會添加最大重試限制以及指數延遲，但是您了解了。

時間耦合–客戶端和服務器無法滿足

我們為正確執行重試所做的努力是客戶端和服務器之間模糊的時間耦合的標志-它們必須同時生活在一起。從技術上講，這不是必需的。想象玩家在48小時內向客戶服務發送一封包含訂單的電子郵件，他們手動更改了庫存。同樣的情況也適用于我們的情況，但是用某種消息代理（例如JMS）替換電子郵件服務器：

@Bean ActiveMQConnectionFactory activeMQConnectionFactory() {return new ActiveMQConnectionFactory("tcp://localhost:61616"); }@Bean JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {return new JmsTemplate(connectionFactory); }

建立ActiveMQ連接后，我們可以簡單地將購買請求發送給經紀人：

private UUID purchaseSword(UUID playerId) {final Sword sword = new Sword(playerId);jmsTemplate.send("purchases", session -> {TextMessage textMessage = session.createTextMessage();textMessage.setText(sword.toJson());return textMessage;});return sword.getUuid(); }

通過用JMS主題上的消息傳遞完全替換同步請求-響應協議，我們暫時將客戶端與服務器分離。他們不再需要同時生活。此外，不止一個生產者和消費者可以相互交流。例如，您可以有多個購買渠道，更重要的是：多個利益相關方，而不僅僅是InventoryService 。更好的是，如果您使用像Kafka這樣的專用消息傳遞系統，則從技術上講，您可以保留數天（數月）的消息而不會降低性能。好處是，如果將另一個購買事件的使用者添加到InventoryService旁邊的系統，它將立即收到許多歷史數據。而且，現在您的應用程序在時間上與代理耦合，因此，由于Kafka是分布式和復制的，因此在這種情況下它可以更好地工作。

異步消息傳遞的缺點

在ReST，SOAP或任何形式的RPC中使用的同步數據交換很容易理解和實現。從延遲的角度來看，誰在乎這種抽象會瘋狂地泄漏（本地方法調用通常比遠程方法快幾個數量級，更不用說它可能因本地未知的眾多原因而失敗），因此開發起來很快。消息傳遞的一個真正警告是反饋渠道。因為沒有響應管道，所以您可以不再只是“ 發送 ”（“ return ”）消息而已。您要么需要帶有一些相關性ID的響應隊列，要么需要每個請求臨時的一次性響應隊列。我們還撒謊了一點，聲稱在兩個系統之間放置消息代理可修復時間耦合。確實如此，但是現在我們耦合到了消息傳遞總線，它也可能會崩潰，特別是因為它通常處于高負載下，有時無法正確復制。

本文展示了在分布式系統中提供保證的一些挑戰和部分解決方案。但是，歸根結底，請記住，“ 僅一次 ”語義幾乎不可能輕松實現，因此仔細檢查您確實需要它們。

翻譯自: https://www.javacodegeeks.com/2015/02/journey-to-idempotency-and-temporal-decoupling.html