前言

在《使用Akka持久化——持久化與快照》一文中介紹瞭如何使用Akka持久化訊息及生成快照。對於叢集應用來說，傳送者發出訊息，只有當收到了接受者的成功回覆才應當認為是一次完整的請求和應答（一些RPC框架只提供了遠端呼叫、序列化/反序列化的機制，但是具體呼叫的成功與否實際是拋給了開發者本人），利用Akka的應答機制很容易實現這些功能。特殊情況下發送者傳送了訊息，但是最終接受者卻沒有接收到訊息，導致這一情況發生的因素很多（例如：傳送者呼叫完傳送介面後，傳送者所在程序奔潰了；網路故障；接收者不存在等）。如果這些訊息的成功接收與處理對於整個應用而言具有強一致性的要求，那麼這些都會導致很多困擾，好在我們可以使用Akka的持久化機制。

傳送者在傳送訊息之前先對訊息進行持久化，那麼無論任何原因導致沒有收到接收者的成功回覆時，我們總能有辦法從持久化資訊中找出那些未被成功回覆的訊息進行重發（這說明接收者接到的訊息有可能會重複，此時需要保證接收者的實現是冥等的）。當接收者收到訊息進行處理後需要向傳送者傳送成功回覆，傳送者收到回覆後的第一個動作應當是對回覆內容的持久化，否則有可能在還未真正對成功回覆處理時宕機或程序奔潰導致回覆訊息丟失（在持久化與收到訊息之間仍然會存在宕機和程序奔潰的情況，只不過這個時間非常短，因此丟失回覆的可能會很低），當持久化回覆訊息完成後，可以自己慢慢來處理這些確認資訊，而不用擔心它們丟失了。

本文將根據Akka官網的例子，對其做一些適應性改造後，向大家展示Akka持久化的另一個強大武器——At least once delivery！

訊息投遞規則

一般而言，訊息投遞有下面三種情況：

• at-most-once 意味著每條應用了這種機制的訊息會被投遞0次或1次。可以說這條訊息可能會丟失。
• at-least-once 意味著每條應用了這種機制的訊息潛在的存在多次投遞嘗試並保證至少會成功一次。就是說這條訊息可能會重複但是不會丟失。
• exactly-once 意味著每條應用了這種機制的訊息只會向接收者準確的傳送一次。換言之，這種訊息既不會丟失也不會重複。

at-most-once的成本最低且效能最高，因為它在傳送完訊息後不會嘗試去記錄任何狀態，然後這條訊息將被他拋之腦後。at-least-once需要傳送者必須認識它所傳送過的訊息，並對沒有收到回覆的訊息進行傳送重試。exactly-once的成本是三者中最高的而且效能卻又是三者中最差的，它除了要求傳送者有記憶和重試能力，還要求接收者能夠認識接收過的訊息並能過濾出那些重複的訊息投遞。Akka的Actor模型一般提供的訊息都屬於at-most-once，那是因為大多數場景都不需要有狀態的訊息投遞，例如web伺服器請求。當你有強一致性需求時，才應該啟用Akka的at-least-once機制，那就是你的Actor不再繼承自UntypedActor，而是繼承自UntypedPersistentActorWithAtLeastOnceDelivery。

配置

如果要使用，那麼需要在中增加以下的一些配置：

    at-least-once-delivery {
      redeliver-interval = 20000
      redelivery-burst-limit = 100
    }

redeliver-interval用於配置重新進行投遞嘗試的時間間隔，單位是毫秒。redelivery-burst-limit用於配置每次重新執行投遞嘗試時傳送的最大訊息條數。

一致性訊息例子

我們首先來看看本例中用到的訊息體MsgSent、Msg、Confirm及MsgConfirmed。MsgSent代表將要傳送的訊息，但是隻用於持久化，持久化完成後會將MsgSent轉換為Msg進行傳送。也就是說Msg才會被真正用於訊息傳送。接收者收到Msg訊息後將向傳送者回復Confirm訊息，需要注意的是Msg和Confirm都有屬性deliveryId，此deliveryId由傳送者的持久化功能生成，一條Msg訊息和其對應的Confirm回覆的deliveryId必須一致，否則在利用UntypedPersistentActorWithAtLeastOnceDelivery對回覆訊息進行確認時會產生嚴重的bug。傳送者收到接收者的Confirm回覆後首先將其轉換為MsgConfirmed，然後對MsgConfirmed進行持久化，最後呼叫UntypedPersistentActorWithAtLeastOnceDelivery提供的confirmDelivery方法對回覆進行確認。MsgSent、Msg、Confirm及MsgConfirmed的程式碼實現如下：

public interface Persistence {

	public static class Msg implements Serializable {
		private static final long serialVersionUID = 1L;
		public final long deliveryId;
		public final String s;

		public Msg(long deliveryId, String s) {
			this.deliveryId = deliveryId;
			this.s = s;
		}
	}

	public static class Confirm implements Serializable {
		private static final long serialVersionUID = 1L;
		public final long deliveryId;

		public Confirm(long deliveryId) {
			this.deliveryId = deliveryId;
		}
	}

	public static class MsgSent implements Serializable {
		private static final long serialVersionUID = 1L;
		public final String s;

		public MsgSent(String s) {
			this.s = s;
		}
	}

	public static class MsgConfirmed implements Serializable {
		private static final long serialVersionUID = 1L;
		public final long deliveryId;

		public MsgConfirmed(long deliveryId) {
			this.deliveryId = deliveryId;
		}
	}

}

服務端

本例中的服務端非常簡單，是一個接收處理Msg訊息，並向傳送者回復Confirm訊息的Actor，程式碼如下:

@Named("MyDestination")
@Scope("prototype")
public class MyDestination extends UntypedActor {
	
	LoggingAdapter log = Logging.getLogger(getContext().system(), this);
	
	public void onReceive(Object message) throws Exception {
		if (message instanceof Msg) {
			Msg msg = (Msg) message;
			log.info("receive msg : " + msg.s + ", deliveryId : " + msg.deliveryId);
			getSender().tell(new Confirm(msg.deliveryId), getSelf());
		} else {
			unhandled(message);
		}
	}
}

服務端的啟動程式碼如下：

		logger.info("Start myDestination");
		final ActorRef myDestination = actorSystem.actorOf(springExt.props("MyDestination"), "myDestination");
		logger.info("Started myDestination");

客戶端

具體介紹客戶端之前，先來列出其實現，程式碼如下：

@Named("MyPersistentActor")
@Scope("prototype")
public class MyPersistentActor extends UntypedPersistentActorWithAtLeastOnceDelivery {
	
	LoggingAdapter log = Logging.getLogger(getContext().system(), this);
	
	private final ActorSelection destination;

	@Override
	public String persistenceId() {
		return "persistence-id";
	}

	public MyPersistentActor(ActorSelection destination) {
		this.destination = destination;
	}

	@Override
	public void onReceiveCommand(Object message) {
		if (message instanceof String) {
			String s = (String) message;
			log.info("receive msg : " + s);
			persist(new MsgSent(s), new Procedure<MsgSent>() {
				public void apply(MsgSent evt) {
					updateState(evt);
				}
			});
		} else if (message instanceof Confirm) {
			Confirm confirm = (Confirm) message;
			log.info("receive confirm with deliveryId : " + confirm.deliveryId);
			persist(new MsgConfirmed(confirm.deliveryId), new Procedure<MsgConfirmed>() {
				public void apply(MsgConfirmed evt) {
					updateState(evt);
				}
			});
		} else if (message instanceof UnconfirmedWarning) {
			log.info("receive unconfirmed warning : " + message);
			// After a number of delivery attempts a AtLeastOnceDelivery.UnconfirmedWarning message will be sent to self. The re-sending will still continue, but you can choose to call confirmDelivery to cancel the re-sending. 
			List<UnconfirmedDelivery> list = ((UnconfirmedWarning) message).getUnconfirmedDeliveries();
			for (UnconfirmedDelivery unconfirmedDelivery : list) {
				Msg msg = (Msg) unconfirmedDelivery.getMessage();
				confirmDelivery(msg.deliveryId);
			}
		} else {
			unhandled(message);
		}
	}

	@Override
	public void onReceiveRecover(Object event) {
		updateState(event);
	}

	void updateState(Object event) {
		if (event instanceof MsgSent) {
			final MsgSent evt = (MsgSent) event;
			deliver(destination, new Function<Long, Object>() {
				public Object apply(Long deliveryId) {
					return new Msg(deliveryId, evt.s);
				}
			});
		} else if (event instanceof MsgConfirmed) {
			final MsgConfirmed evt = (MsgConfirmed) event;
			confirmDelivery(evt.deliveryId);
		}
	}
}

正如我們之前所述——要使用at-least-once的能力，就必須繼承UntypedPersistentActorWithAtLeastOnceDelivery。有關MsgSent、Msg、Confirm及MsgConfirmed等訊息的處理過程已經介紹過，這裡不再贅述。我們注意到onReceiveCommand方法還處理了一種名為UnconfirmedWarning的訊息，這類訊息將在at-least-once機制下進行無限或者一定數量的投遞嘗試後傳送給當前Actor，這裡的數量可以通過在at-least-once-delivery配置中增加配置項warn-after-number-of-unconfirmed-attempts來調整，例如：

    at-least-once-delivery {
      redeliver-interval = 20000
      redelivery-burst-limit = 100
      warn-after-number-of-unconfirmed-attempts = 6
    }

當你收到UnconfirmedWarning的訊息時，說明已經超出了你期望的最大重試次數，此時可以做一些控制了，例如：對於這些訊息傳送報警、丟棄等。本例中選擇了丟棄。

UntypedPersistentActorWithAtLeastOnceDelivery的狀態由那些尚未被確認的訊息和一個序列號組成。UntypedPersistentActorWithAtLeastOnceDelivery本身不會儲存這些狀態，依然需要你在呼叫deliver方法投遞訊息之前，呼叫persist方法持久化這些事件或訊息，以便於當持久化Actor能夠在恢復階段恢復。在恢復階段，deliver方法並不會將發出訊息，此時持久化Actor一面恢復，一面只能等待接收回復。當恢復完成，deliver將傳送那些被快取的訊息（除了收到回覆，並呼叫confirmDelivery方法的訊息）。

執行例子

本文將率先啟動客戶端並向服務端傳送hello-1，hello-2，hello-3這三訊息，但是由於服務端此時並未啟動，所以客戶端會不斷重試，直到重試達到上限或者受到回覆並確認。服務端傳送訊息的程式碼如下：

		logger.info("Start myPersistentActor");
		final String path = "akka.tcp://[email protected]:2551/user/myDestination";
		final ActorSelection destination = actorSystem.actorSelection(path);
		final ActorRef myPersistentActor = actorSystem.actorOf(springExt.props("MyPersistentActor", destination), "myPersistentActor");
		actorMap.put("myPersistentActor", myPersistentActor);
		logger.info("Started myPersistentActor");
		myPersistentActor.tell("hello-1", null);
		myPersistentActor.tell("hello-2", null);
		myPersistentActor.tell("hello-3", null);

客戶端傳送三條訊息後，日誌中立馬打印出了以下內容：

但是一直未受到回覆資訊，然後我們啟動服務端，不一會就看到了以下日誌輸出：

我們再來看看客戶端，發現已經收到了回覆，內容如下：

總結

通過使用UntypedPersistentActorWithAtLeastOnceDelivery提供的persist、deliver及confirmDelivery等方法可以對整個應用的at-least-once需求，輕鬆實現在框架層面上一致的實現。

其它Akka應用的博文如下：

後記：個人總結整理的《深入理解Spark：核心思想與原始碼分析》一書現在已經正式出版上市，目前京東、噹噹、天貓等網站均有銷售，歡迎感興趣的同學購買。