gpt4 book ai didi

java - 如何在 Zookeepr 中模拟获取锁的 void 方法,并且我们还可以验证锁

转载 作者:行者123 更新时间:2023-12-01 22:29:42 25 4
gpt4 key购买 nike

我必须为使用 Zookeeper 获取锁并使用 CompletableFuture 处理数据的方法创建一个单元测试用例

下面是高级代码:

import lombok.Data;
@Data
public class ConfigurationsIntegrationModel {
public enum InteractionType {
TEST,
DEV;
}
private InteractionType interactionType;
private String lockName;
}

import org.springframework.stereotype.Service;
import java.util.Arrays;
import java.util.List;

@Service("configurationsIntegrationService")
public interface ConfigurationsIntegrationService {
public default List<ConfigurationsIntegrationModel> getRecords(ConfigurationsIntegrationModel.InteractionType integrationType) {
return Arrays.asList(getDynamicIntegrationConfigurationMock(integrationType));
}
private static ConfigurationsIntegrationModel getDynamicIntegrationConfigurationMock(ConfigurationsIntegrationModel.InteractionType integrationType) {
ConfigurationsIntegrationModel configurationsIntegration = new ConfigurationsIntegrationModel();
configurationsIntegration.setLockName("Test_Lock");
configurationsIntegration.setInteractionType(integrationType);
return configurationsIntegration;
}
}

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
import org.apache.curator.retry.RetryNTimes;
import java.util.concurrent.TimeUnit;
public class DistributedLockProcessor {
private CuratorFramework client;
private String path;
public DistributedLockProcessor(String host, String path) {
RetryPolicy retryPolicy = new RetryNTimes(5, 90);
client = CuratorFrameworkFactory.newClient(host, retryPolicy);
client.start();
}
public InterProcessLock acquire(String lockName) throws Exception {
InterProcessSemaphoreMutex sharedLock = new InterProcessSemaphoreMutex(client, path + "/" + lockName);
if (!sharedLock.acquire(0, TimeUnit.SECONDS)) {
return null;
}
return sharedLock;
}
public boolean release(InterProcessLock sharedLock) throws Exception {
sharedLock.release();
return true;
}
}

import org.apache.curator.framework.recipes.locks.InterProcessLock;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
public class LockingExecutorProcessor<T> {
private Executor executor = null;
private DistributedLockProcessor distributedLock = null;
public LockingExecutorProcessor(String host, String path, int executorCount) {
executor = Executors.newFixedThreadPool(executorCount);
distributedLock = new DistributedLockProcessor(host, path);
}
public void process(List<String> locks, List<T> items, Consumer<T> consumer) throws ExecutionException, InterruptedException {
final List<CompletableFuture<Void>> completableFutures = new ArrayList<>();
for (int i = 0; i < locks.size(); i++) {
final int record = i;
CompletableFuture<Void> future =
CompletableFuture.runAsync(
() -> {
InterProcessLock interProcessLock = null;
try {
interProcessLock = distributedLock.acquire(locks.get(record));
} catch (Exception e) {
e.printStackTrace();
}
if (interProcessLock != null) {
consumer.accept(items.get(record));
}
}, executor);
completableFutures.add(future);
}
CompletableFuture<Void> completable = CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()]));
completable.get();
}
}

import org.springframework.stereotype.Service;
import java.util.Arrays;
import java.util.List;
@Service("messageService")
public interface MessageService {
public default List<String> getMessagesList(ConfigurationsIntegrationModel.InteractionType integrationType) {
return Arrays.asList("Message1", "Message2", "Message3","Message4");
}
}

import com.google.common.util.concurrent.RateLimiter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
@Component("sampleJob")
public class SampleJob {
@Autowired
private LockingExecutorProcessor<ConfigurationsIntegrationModel> lockingExecutorProcessor;
@Autowired
private ConfigurationsIntegrationService configurationsIntegrationService;
@Autowired
private RateLimiter rateLimiter;
@Autowired
private MessageService messageService;

private List<String> getLockNames(List<ConfigurationsIntegrationModel> integrationConfigurations) {
List<String> lockNames = new ArrayList<>();
for (ConfigurationsIntegrationModel integrationConfiguration : integrationConfigurations) {
lockNames.add(integrationConfiguration.getLockName());
}
return lockNames;
}
@Scheduled(fixedRateString = "100")
public void executeJob() throws ExecutionException, InterruptedException {
List<ConfigurationsIntegrationModel> testRecords = configurationsIntegrationService.getRecords(ConfigurationsIntegrationModel.InteractionType.TEST);
List<String> lockNames = getLockNames(testRecords);
lockingExecutorProcessor.process(
lockNames,
testRecords,
recordsConfig -> {
final List<CompletableFuture<Void>> completableFutures = new ArrayList<>();
List<String> msgList = messageService.getMessagesList(recordsConfig.getInteractionType());
for (String message : msgList) {
completableFutures.add(
CompletableFuture.runAsync(
() -> {
System.out.println("Message is @@@@ "+ message);
}, Executors.newFixedThreadPool(10)));
}
});
}
}

下面是我到目前为止尝试过的测试用例:

import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
import org.mockito.junit.MockitoJUnitRunner;
import java.util.concurrent.ExecutionException;
@RunWith(MockitoJUnitRunner.class)
public class SampleJobTest {
@InjectMocks
private SampleJob sampleJob = new SampleJob();
@Mock
private ConfigurationsIntegrationService configurationsIntegrationService;
@Mock
private MessageService messageService;
@Mock
private LockingExecutorProcessor<ConfigurationsIntegrationModel> lockingExecutorProcessor;

@Test
public void testSampleJob() throws ExecutionException, InterruptedException {
Mockito.doCallRealMethod().when(lockingExecutorProcessor).process(Mockito.any(), Mockito.any(), Mockito.any());
Mockito.doCallRealMethod().when(configurationsIntegrationService).getRecords(Mockito.any());
Mockito.doCallRealMethod().when(messageService).getMessagesList(Mockito.any());
sampleJob.executeJob();
}
}

当我调试代码时,它会在 LockingExecutorProcessor 的 CompletableFuture.runAsync 行处中断并抛出空指针错误;原因是distributedLockProcessor对象为null;

我们如何模拟它以及如何连接到测试zookeeper服务器而不是原始服务器以确保锁定正常工作

最佳答案

这样就完成了。现在测试用例运行良好。我使用了 curator-test 依赖项,还使用了反射来模拟私有(private)对象。

关于java - 如何在 Zookeepr 中模拟获取锁的 void 方法,并且我们还可以验证锁,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/58553848/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com