[Spring AOP] Consumer onMessage() 비동기 수신 로깅

■ AOP를 통해 onMessage() 비동기 수신 로깅

 
Redis가 Consume할 때 호출되어 작업을 처리하기 시작하는 부분인 onMessage() 메서드를 AOP를 통해 로깅하려는 상황이었다. 기능 구현을 위한 샘플 코드를 짜서 테스트해보았다.
 
구조는 이러하다.
- ConsumerFactory: 메시지를 구독할 컨슈머들을 생성
- StreamListenerImpl: 구독한 메세지를 처리할 onMessage 구현
- ListenContainer: 위 리스너를 관리할 컨테이너 설정
- LoggingAspect: 테스트 대상인 Aspect들
- RedisConfig: 기본 레디스 설정
 
 
*참고: ListenerContainer, Listener, 그리고 Consumer 객체간의 관계를 이해하는 것이 필요하다면, 지난 글에 정리해두었다. https://splendidlolli.tistory.com/775

[Redis] StreamListener(onMessage)와 Consumer 객체간 관계

 

---

■ 테스트할 Aspect

onMessage 로깅 목적으로 Advice중 @Before, @AfterReturning, @Around를 모두 사용할 예정이어서 일단 이런 Aspect를 작성했다.
 
LoggingAsepct.java

@Aspect
@Component
public class LoggingAspect {
    @Before("execution(* com.example.demo.StreamListenerImpl.onMessage(..))")
    public void logging1() {
        System.out.println("aspect: @Before");
    }
    @AfterReturning("execution(* com.example.demo.StreamListenerImpl.onMessage(..))")
    public void logging2() {
        System.out.println("aspect: @AfterReturning");
    }
    
    @Around("execution(* com.example.demo.StreamListenerImpl.onMessage(..))")
    public void logging3(ProceedingJoinPoint joinpoint) throws Throwable {
        System.out.println("aspect: @Around: start");
        joinpoint.proceed();
        System.out.println("aspect: @Around: end");
    }
}

 
이 세개의 Aspect를 모두 onMessage를 동작하게끔 한다면 아래와 같은 순서가 된다.

aspect: @Around: start
aspect: @Before
~~ StreamListenerImpl onMessage ~~
aspect: @AfterReturning
aspect: @Around: end

 
 
 

---

■ 필요한 Bean 설정: Listener

AOP가 발현되는 대상은 Spring Bean으로 관리되는 대상이다.
그래서 중요한 점은 Listener가 Bean으로 잘 등록되어야 한다는 것이다. Listener의 onMessage가 AOP의 대상이 되도록 설정을 잘 해야 한다.
 
Bean 관련해서 간단한 테스트를 해보았다.
비동기 수신 메서드인 onMessage에 AOP를 적용하기 위해서 반드시 Bean이어야 하는 것은 무엇일까? 궁금했던 후보는 ListenerContainer, Listener, Consumer다. 결론은 onMessage가 구현되어있는 Listener만 Bean이면 AOP 적용이 됐다.
 
 
 

ⓐ Contiainer와 Listener 모두 Bean 등록 되어있는 상태
Aspect 작동 함
 
ⓑ  Container는 Bean 아니고, Listener만 Bean인 상태
Aspect 작동 함
 
ⓒ 둘다 Bean이 아닌 경우
작동 안함
 
ⓓ Container가 Bean인데 Listener가 Bean 아닐 경우
작동 안함

 

---

■ AOP 유의점: Proxy

Listener를 설정하는 과정에서 Proxy와 관련해 유의할 점이 있다.
 
ListenerContainer에 Listener 객체를 등록하는 코드는 다양하겠지만 예를 두개 들면 이런식으로 넣어주게된다. 
- container.receiveAutoAck(consumer, streamOffset, listener)
- containerProperties.setMessageListener(listener);
 
플젝 코드에서 AOP가 동작을 안하길래 확인 해봤더니 listener 부분을 this로 주입하고 있었다. 즉 Listener 구현 클래스 내부에서 초기화를 하면서 자기자신을 this로 넣은 상황이다.
이 상황은 "프록시 내부 호출" 이라는 키워드로 검색해보면 참고가 될 것이다.
(예시: https://m.blog.naver.com/fbfbf1/223225345341)
 
그렇게 this로 넣으면 AOP의 대상이 되지 못한다. AOP는 Spring에서 프록시 패턴으로 구현되었기 때문에 프록시 객체를 주입해야 AOP 기능이 적용된다. 그러나 this를 주입하게되면 프록시 Listener 객체의 onMessage가 아닌, 실제 Listener 객체의 onMessage를 호출하는 거라서 AOP가 작동하지 못한다. 
 
*참고: 이와 같은 문제로 stackovelflow의 질문글도 하나 찾았다: https://stackoverflow.com/questions/51028269/dynamic-kafka-consumer-with-aop

Dynamic Kafka Consumer with AOP

 
 
핵심은 프록시를 주입해주는 것이다. 
 
1. Factory, Config 등을 별도로 두고 DI해주기
- 순환 고리를 끊어줄 방법을 고민하자는 것이다.
- 구조 자체를 이렇게 변경해주는 것이 권장되는 방법이다.
 
2. Listener 자기자신 타입 private 필드를 Setter를 통해 Autowired받아 this 대신 사용하기
- 생성자 주입은 안된다. 순환 참조 문제가 발생한다.
- 그러나 본 Setter 방법도 Spring 2.x 이상에서는 순환 참조 문제가 발생한다. 최후 수단으로 spring.main.allow-circular-references=true 설정으로 순환참조를 허용해줄 수 있다.
 
 
* 참고: 이렇게 해도 안된다면 점검해볼만한 것이.... 혹시 proxy의 proxy를 리스너로 등록해주는 상황은 없는지 검토해보자. 내가 건드린 클래스는, List 속에 리스너를 넣어두고 그 List에서 리스너를 꺼내 consumer로 등록해주는 뭔가 이중적인 구조의 코드였다. 그런데 List 속에 넣은 것도 proxy라서 onMessage가 아예 동작을 안했다. List에는 this를, 그리고 컨슈머로 등록해줄 때는 proxy를 넣어줘서 고쳤다.
 
 

---

끝!