鸟瞰 Java 并发框架(4)

Disruptor 框架在下列场合性能更好：与事件驱动的体系结构一起使用，或主要关注内存任务的单个生产者和多个消费者。

static { 
    int userId = new Random().nextInt(10) + 1; 
 
    // 示例 Event-Handler; count down latch 用于使线程与 http 线程同步 
    EventHandler<Event> postsApiHandler = (event, sequence, endOfBatch) -> { 
        event.posts = JsonService.getPosts(); 
        event.countDownLatch.countDown(); 
    }; 
 
    // 配置 Disputor 用于处理事件 
    DISRUPTOR.handleEventsWith(postsApiHandler, commentsApiHandler, albumsApiHandler) 
    .handleEventsWithWorkerPool(photosApiHandler1, photosApiHandler2) 
    .thenHandleEventsWithWorkerPool(postsAndCommentsResponseHandler1, postsAndCommentsResponseHandler2) 
    .handleEventsWithWorkerPool(albumsAndPhotosResponseHandler1, albumsAndPhotosResponseHandler2); 
    DISRUPTOR.start(); 
} 
 
// 对于每个请求，在 RingBuffer 中发布一个事件: 
Event event = null; 
RingBuffer<Event> ringBuffer = DISRUPTOR.getRingBuffer(); 
long sequence = ringBuffer.next(); 
CountDownLatch countDownLatch = new CountDownLatch(6); 
try { 
    event = ringBuffer.get(sequence); 
    event.countDownLatch = countDownLatch; 
    event.startTime = System.currentTimeMillis(); 
} finally { 
    ringBuffer.publish(sequence); 
} 
try { 
    event.countDownLatch.await(); 
} catch (InterruptedException e) { 
    e.printStackTrace(); 
} 

10. Akka

图片来自：https://blog.codecentric.de/en/2015/08/introduction-to-akka-actors/

• Akka 库的主要优势在于它拥有构建分布式系统的本地支持。
• 它运行在一个叫做 Actor System 的系统上。这个系统抽象了线程的概念，Actor System 中的 Actor 通过异步消息进行通信，这类似于生产者和消费者之间的通信。
• 这种额外的抽象级别有助于 Actor System 提供诸如容错、位置透明等特性。
• 使用正确的 Actor-to-Thread 策略，可以对该框架进行优化，使其性能优于上表所示的结果。虽然它不能在单个节点上与传统方法的性能匹敌，但是由于其构建分布式和弹性系统的能力，仍然是首选。

10.1 示例代码

// 来自 controller : 
Actors.masterActor.tell(new Master.Request("Get Response", event, Actors.workerActor), ActorRef.noSender()); 
 
// handler : 
public Receive createReceive() { 
    return receiveBuilder().match(Request.class, request -> { 
    Event event = request.event; // Ideally, immutable data structures should be used here. 
    request.worker.tell(new JsonServiceWorker.Request("posts", event), getSelf()); 
    request.worker.tell(new JsonServiceWorker.Request("comments", event), getSelf()); 
    request.worker.tell(new JsonServiceWorker.Request("albums", event), getSelf()); 
    request.worker.tell(new JsonServiceWorker.Request("photos", event), getSelf()); 
    }).match(Event.class, e -> { 
    if (e.posts != null && e.comments != null & e.albums != null & e.photos != null) { 
    int userId = new Random().nextInt(10) + 1; 
    String postsAndCommentsOfRandomUser = ResponseUtil.getPostsAndCommentsOfRandomUser(userId, e.posts, 
    e.comments); 
    String albumsAndPhotosOfRandomUser = ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, e.albums, 
    e.photos); 
    String response = postsAndCommentsOfRandomUser + albumsAndPhotosOfRandomUser; 
    e.response = response; 
    e.countDownLatch.countDown(); 
    } 
    }).build(); 
} 

（编辑：ASP站长网）