java - 通过异步执行程序使用restTemplate时，带有MockRestService的ConcurrentModificationException

尝试用spring的MockRestService模拟出剩余响应以进行集成测试，但是当实际代码异步使用rest模板时，AbstractRequestExpectationManager会一直运行到ConcurrentModificationException中。

测试伪代码段:

@Autowired
RestTemplate restTemplate;
MockRestServiceServer mockRestServiceServer;

@Test
public test() {
    // given
    mockRestServiceServer = MockRestServiceServer
            .bindTo( restTemplate )
            .ignoreExpectOrder()  // supported from spring 4.3
            .build();
    prepareRestResponse( "/resource/url", "mock json content".getBytes() );
    // when
    myservice.refreshPricesForProductGroup( 2 );

    // then
    // assertions
}

private void prepareRestResponse( final String urlTail, final byte[] responseContent ) {
    mockRestServiceServer
            .expect( requestTo( endsWith( urlTail ) ) )
            .andExpect( method( HttpMethod.GET ) )
            .andRespond( withSuccess()
                    .body( responseContent )
                    .contentType( APPLICATION_JSON_UTF8 ) );
}

实际代码访问其余模板:

@Autowired
Executor executor
@Autowired
PriceRestClient priceClient
@Autowired
ProductRestClient productClient

/../

private void refreshPricesForProductGroup( final int groupId ) {

    List<Product> products = productClient.findAllProductsForGroup( groupId );

    products.forEach( p ->
            executor.execute( () -> {
                final Price price = priceClient.getPrice( p.getId() );
                priceRepository.updatePrice( price );
            } )
    );
}

PriceRestClient.getPrice()执行简单的调用:

Price getPrice( String productId ) {

    try {
        ResponseEntity<byte[]> entity = restTemplate.exchange(
                restUtil.getProductPriceDataUrl(),
                HttpMethod.GET,
                restUtil.createGzipEncodingRequestEntity(),
                byte[].class,
                productId );

        if ( entity.getStatusCode() == HttpStatus.OK ) {
            String body = restUtil.unmarshalGzipBody( entity.getBody() );
            return priceEntityParser.parse( body );
        }

    } catch ( HttpClientErrorException e ) {
        // TODO
    } catch ( ResourceAccessException e ) {
        // TODO
    } catch ( IOException e ) {
        // TODO
    }

    return null;
}

引发异常:

Exception in thread "AsyncExecutor-2" java.util.ConcurrentModificationException
    at java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:711)
    at java.util.LinkedHashMap$LinkedKeyIterator.next(LinkedHashMap.java:734)
    at org.springframework.test.web.client.AbstractRequestExpectationManager$RequestExpectationGroup.findExpectation(AbstractRequestExpectationManager.java:167)
    at org.springframework.test.web.client.UnorderedRequestExpectationManager.validateRequestInternal(UnorderedRequestExpectationManager.java:42)
    at org.springframework.test.web.client.AbstractRequestExpectationManager.validateRequest(AbstractRequestExpectationManager.java:71)
    at org.springframework.test.web.client.MockRestServiceServer$MockClientHttpRequestFactory$1.executeInternal(MockRestServiceServer.java:286)
    at org.springframework.mock.http.client.MockClientHttpRequest.execute(MockClientHttpRequest.java:93)
    at org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:93)
    at com.mycompany.myproduct.web.client.HttpRequestInterceptorLoggingClient.interceptReq(HttpRequestInterceptorLoggingClient.java:32)
    at org.springframework.http.client.InterceptingClientHttpRequest$InterceptingRequestExecution.execute(InterceptingClientHttpRequest.java:85)
    at org.springframework.http.client.InterceptingClientHttpRequest.executeInternal(InterceptingClientHttpRequest.java:69)
    at org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
    at org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
    at org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:596)
    at org.springframework.web.client.RestTemplate.execute(RestTemplate.java:557)
    at org.springframework.web.client.RestTemplate.exchange(RestTemplate.java:475)
    at com.mycompany.myproduct.rest.PriceRestClient.getPrice(PriceRestClient.java:48)
    at com.mycompany.myproduct.service.ProductPriceSourcingService.lambda$null$29(ProductPriceSourcingService.java:132)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

我在这里做错什么了吗，或者可能是MockRestService的错误？

最佳答案

我只需要线程安全性，因此我稍微重构了laur's answer并删除了其他功能(即时修改期望)。留在这里供参考。

import org.springframework.http.client.ClientHttpRequest;
import org.springframework.http.client.ClientHttpResponse;
import org.springframework.test.web.client.AbstractRequestExpectationManager;
import org.springframework.test.web.client.RequestExpectation;
import org.springframework.test.web.client.UnorderedRequestExpectationManager;

import java.io.IOException;
import java.util.*;

public class SynchronizedUnorderedRequestExpectationManager extends AbstractRequestExpectationManager {
    private final List<RequestExpectation> expectations = Collections.synchronizedList(new LinkedList());
    private final List<ClientHttpRequest> requests = Collections.synchronizedList(new LinkedList());
    private final SynchronizedRequestExpectationGroup remainingExpectations = new SynchronizedRequestExpectationGroup();

    @Override
    protected List<RequestExpectation> getExpectations() {
        return this.expectations;
    }

    @Override
    protected List<ClientHttpRequest> getRequests() {
        return this.requests;
    }

    protected void afterExpectationsDeclared() {
        this.remainingExpectations.updateAll(this.getExpectations());
    }

    public ClientHttpResponse validateRequestInternal(ClientHttpRequest request) throws IOException {
        RequestExpectation expectation = this.remainingExpectations.findExpectation(request);

        if (expectation != null) {
            ClientHttpResponse response = expectation.createResponse(request);
            this.remainingExpectations.update(expectation);
            return response;
        }

        throw this.createUnexpectedRequestError(request);
    }

    public void reset() {
        super.reset();
        this.remainingExpectations.reset();
    }

    protected static class SynchronizedRequestExpectationGroup extends RequestExpectationGroup {

        private final Set<RequestExpectation> expectations = Collections.synchronizedSet(new LinkedHashSet<>());

        @Override
        public Set<RequestExpectation> getExpectations() {
            return this.expectations;
        }

        @Override
        public void updateAll(Collection<RequestExpectation> expectations) {
            synchronized (expectations) {
                super.updateAll(expectations);
            }
        }

        @Override
        public RequestExpectation findExpectation(ClientHttpRequest request) throws IOException {
            synchronized (expectations) {
                return super.findExpectation(request);
            }
        }
    }
}

背景

该代码基本上是从UnorderedRequestExpectationManager复制的。不幸的是，这仍然是必需的，因为原始类与不安全线程的AbstractRequestExpectationManager.RequestExpectationGroup紧密耦合。为了替换依赖关系，需要重写此类。其他线程不安全的集合依赖项(来自AbstractRequestExpectationManager)被重写getExpectations和getRequests方法取代。集合迭代由关键部分保护。

用法

RestTemplate restTemplate = new RestTemplate();
MockRestServiceServer mockServer = MockRestServiceServer.bindTo(restTemplate)
    .build(new SynchronizedUnorderedRequestExpectationManager());

谢谢劳尔! (已投票)