java - Akka Actor 和 future : Understanding by example

Question

我正在尝试学习 Akka Actor 和 future ，但在阅读了 http://akka.io 上的文档之后做http://doc.akka.io/docs/akka/2.0.2/intro/getting-started-first-java.html我仍然有一些理解问题。我猜计算 Pi 的值很多人也可以联系起来，但不是我=)。我搜索了一下但还没有找到任何适合我的例子。因此，我想我会拿一些我的真实代码并把它扔在这里，并用它来交换一个如何用 Akka 做到这一点的例子。

好的，我们开始吧:

我有一个 java play2 应用程序，我需要从我的数据库中获取一些数据并将其索引到我的 elasticsearch 实例中。

1. 我调用数据库并获取 field 的 ID。

2. 然后我拆分列表并创建几个可调用的索引任务。

3. 之后，我调用所有任务，其中每个任务为分配的 ID 收集场所来自数据库。

4. 对于每个 field ，将其索引到 elasticsearch 实例并使其可搜索。

5. 完成。

Application.java:

public class Application extends Controller {

  private static final int VENUE_BATCH = 1000;
  private static int size;

  public static Result index() {

      List<Long> venueIds = DbService.getAllVenueIds();
      size = venueIds.size();
      Logger.info("Will index " + size + " items in total.");
      ExecutorService service = Executors.newFixedThreadPool(getRuntime().availableProcessors());
      int startIx = 0;
      Collection<Callable<Object>> indexTasks = new ArrayList<Callable<Object>>();
      do {
          int endIx = Math.min(startIx + VENUE_BATCH, size);
          List<Long> subList = venueIds.subList(startIx, endIx);
          VenueIndexTask indexTask = new VenueIndexTask(subList);
          indexTasks.add(indexTask);
      } while ((startIx += VENUE_BATCH) < size);

      Logger.info("Invoking all tasks!");
      try {
          service.invokeAll(indexTasks);
      } catch (InterruptedException e) {
          e.printStackTrace();
      }

      return ok(index.render("Done indexing."));
  } 
}

地点任务:

public class VenueIndexTask implements Callable<Object> {

    private List<Long> idSubList;

    public VenueIndexTask(List<Long> idSubList){
        this.idSubList = idSubList;
        Logger.debug("Creating task which will index " + idSubList.size() + " items. " +
                "Range: " + rangeAsString() + ".");
    }

    @Override
    public Object call() throws Exception {
        List<Venue> venues = DbService.getVenuesForIds(idSubList);
        Logger.debug("Doing some indexing: "+venues.size());

        for(Venue venue : venues) {
            venue.index();
        }
        return null;
    }
    private String rangeAsString() {
        return "[" + idSubList.get(0) + "-" + idSubList.get(idSubList.size() - 1) + "]";
    }
}

地点:

@IndexType(name = "venue")
public class Venue extends Index {

    private String name;

    // Find method static for request
    public static Finder<Venue> find = new Finder<Venue>(Venue.class);

    public Venue() {
    }

    public Venue(String id, String name) {
        super.id = id;
        this.name = name;
    }

    @Override
    public Map toIndex() {
        HashMap map = new HashMap();
        map.put("id", super.id);
        map.put("name", name);
        return map;
    }

    @Override
    public Indexable fromIndex(Map map) {
        if (map == null) {
            return this;
        }
        this.name = (String) map.get("name");
        return this;
    }
}

所以你们所有的 Akka 人都发疯了!请尽可能多地提出可以使用的酷 future 功能或我可以用来学习这些东西的任何其他知识/代码。

Answer 1

我喜欢将 Akka(或任何其他基于消息的系统)视为像传送带一样思考，就像在工厂中一样。 Actors 的一种简化思维方式可能是订购披萨。

• 你，饥饿的顾客( Actor /角色)向披萨店发送订单(一条消息)

• 客服( Actor /角色)接您的订单，给您订单号( future )

• 如果您不耐烦，您可能已经在电话/互联网/商店等到您拿到披萨(同步/阻塞交易)，否则您会对订单号感到满意，稍后再检查(非阻塞)

• 客服在厨房经理( Actor )的监督下将消息发送给厨师( Actor )。这是一个流程很重的厨房，有层次感。 Akka 喜欢这样。见 Supervision

• 厨师创建一个新的比萨饼，并附上订单的详细信息(一条新消息)，然后通过送货经理(Supervisor Actor)将其传递给送货员(Actor)。

• 在此过程中，您的订单详情没有改变，那将是一场噩梦。如果你想要纯奶酪，如果你有意大利辣香肠，你会不高兴!所有消息都应该是不可变的!但是，对于不同的参与者，消息可能会有所不同。送货员会期待披萨和附上的订单详细信息，厨师会期待订单。当消息需要更改时，会创建一条新消息。

• 每个 Actor 都擅长一个角色，如果一个人必须完成所有任务，效果如何？可能是某些 Actor 的数量超过了其他 Actor (例如，厨师 10 个线程，送货员 2 个线程，客户服务 1 个线程)。

• 阻止行为是一种痛苦，想象一下客户服务正在等待厨师和送货员在看到下一位客户之前？

希望我对您有所帮助，这是一个巨大的话题，也是一个很大的转变。祝你好运