Akka實踐一些總結

最近在一些服務中使用了akka,主要用來作異步解耦和本地消息分發(路由),這裏簡單總結一下用法.

與spring集成

網上有很多集成的例子,要使用到spring的擴展.
我這邊沒有這樣處理,而是簡單把ActorSystem建立的actor的過程放在了spring configuration裏,把ActorRef做爲bean,畢竟actor自己不能爲單例可是ref能夠.

actor要使用一些bean的話就所有都由構造函數傳入.
以下:

public class UserActor extends AbstractLoggingActor {

    private UserService userService;

    public static Props props(UserService userService) {
        return Props.create(UserActor.class, () -> new UserActor(userService));
    }

    public UserActor(UserService userService) {
        this.userService = userService;
    }

在java中使用

使用中能感受和java仍是有些隔閡的...
特別是容器(Collection)之間的轉換和處理.
給出一些例子,多是我沒發現更方便的寫法..

List<Routee> routees = new ArrayList<>();
senders.forEach(e -> {
            try {
                ActorRef ref = getContext().actorOf(createdProps.apply(e), prefix + "-" + e.getId());
                routees.add(new ActorRefRoutee(ref));
            } catch (InvalidActorNameException ex) {
                log().error(ex, "name already in use");
            }
        });
router = new Router(new RoundRobinRoutingLogic(), routees);

這邊原本用一個map就能解決了,但會要求類型強轉Router的第二個參數是java.lang.Iterable<Routee>而不是java.lang.Iterable<? extends Routee>(ActorRefRoutee是其子類),再加上還要處理重名錯誤,仍是用了這種錯誤的forEach使用方式.

private void stopRoutee(Router router) {
        List<Routee> routees = new ArrayList<>(seqAsJavaList(router.routees()));
        routees.forEach(e -> {
            ActorRefRoutee ae = (ActorRefRoutee) e;
            ActorRef ref = ae.ref();
            // 移除這個routee
            router.removeRoutee(ae);
            // 中止這個ref
            context().stop(ref);
        });
    }

router.routees()返回的是個IndexSeq...和java的沒有兼容和對應.
最開始找了一些,後來發現有scala.collection.JavaConversions能夠作兼容...感覺到了隔閡.

此外ask模式,akka爲java提供了PatternsCS.這個類返回的是java的CompletionStage類型相比Patterns,比較友好.

使用誤區

最容易誤用的一點就是消息處理中有阻塞操做,好比直接在actor中進行數據庫操做和處理網絡請求,而使用的actor也沒有和線程綁定,這種狀況須要使用額外的線程池,這個其實和用netty是同樣的狀況,用來進行應用調度的線程數有限.

因爲上面提到的點,因而在actor內可能就會存在一些回調,一不當心就可能直接調用/改變actor內部的狀態(例如在回調直接訪問field).

private List<String> list;
... ...
public Receive createReceive() {
    return receiveBuilder()
            .match(Message.class, e -> xxxx.onSuccess(r -> list.add(r)))
            ... ...

這樣等於內部狀態被多線程訪問,破壞了actor內部的狀態,正確的作法是在回調中給本身發送消息.