Guava AsyncEventBus

AsyncEventBus

 

// AsyncEventBus Configuration

import com.google.common.eventbus.AsyncEventBus;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

private static final int CORE_POOL_SIZE = 80;
private static final int MAX_POOL_SIZE = 100;
private static final int KEEP_ALIVE_TIME = 1;
private static final int CAPACITY = 150000;

@Bean
public AsyncEventBus asyncEventBus() {
	return new AsyncEventBus(limitedAsyncExecutor());
}

private Executor limitedAsyncExecutor() {
	return new ThreadPoolExecutor(
		CORE_POOL_SIZE,
		MAX_POOL_SIZE,
		KEEP_ALIVE_TIME,
		TimeUnit.SECONDS,
		new ArrayBlockingQueue<>(CAPACITY), (r, executor) -> log.warn("fail eventBus processing")
	);
}

---

// AsyncEvent Subscriber

import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.Subscribe;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;

@Slf4j
@Service
public class SomeSubscriber {
	@Autowired
	private AsyncEventBus asyncEventBus;

	@Autowired
	private SomeModifier someModifier;

	@Subscribe
	@AllowConcurrentEvents
	public void consumeSomeEvent(SomeDTO someDTO) {
		try {
			someModifier.modify(someDTO);
		} catch (Exception e) {
			log.warn("modify failed : {}", e);
		}
	}

	@PostConstruct
	private void registerEvent() {
		asyncEventBus.register(this);
	}
}

---

// AsyncEventBus using

@Autowired
private AsyncEventBus asyncEventBus;

asyncEventBus.post(someDto);

---

 

• @Subscribe 메소드는 싱글 쓰레드로 동작한다.
• @Subscribe 에 @AllowConcurrentEvents 애노테이션을 추가하면 멀티 쓰레드로 동작한다.

 

참조 :

- http://kwonnam.pe.kr/wiki/java/guava/eventbus