### Question:
ConsumerServiceImpl extends AbstractBaseService<ConsumerEntity> implements ConsumerService { private void doSaveMsg(PublishMessageRequest request, PublishMessageResponse response, List<QueueEntity> queueEntities, List<Message01Entity> message01Entities) { int tryCount = 0; int queueSize = queueEntities.size(); Exception last = null; Transaction transaction = null; if (request.getSynFlag() == 1) { transaction = Tracer.newTransaction("Publish", request.getTopicName()); } else { transaction = Tracer.newTransaction("Publish-Asyn", request.getTopicName()); } String key = request.getTopicName(); Map<String, AtomicInteger> counterTemp = counter.get(); if (!counterTemp.containsKey(key)) { counterTemp.put(key, new AtomicInteger(0)); } counterTemp.get(key).compareAndSet(Integer.MAX_VALUE, 0); int count = counterTemp.get(key).incrementAndGet(); while (tryCount <= queueSize) { try { QueueEntity temp = queueEntities.get(count % queueEntities.size()); count++; if (!checkFailTime(request.getTopicName(), temp, null)) { continue; } doSaveMsg(message01Entities, request, response, temp); last = null; if (response.isSuc()) { transaction.setStatus(Transaction.SUCCESS); addPublishLog(message01Entities, request, MqConst.INFO, null); } else { transaction.setStatus(response.getMsg()); last = new RuntimeException(response.getMsg()); } break; } catch (Exception e) { tryCount++; response.setSuc(false); response.setMsg("消息保存失败！"); last = e; } } if (last != null) { transaction.setStatus(last); addPublishLog(message01Entities, request, MqConst.ERROR, last); sendPublishFailMail(request, last, 2); } transaction.complete(); }  @Override @Transactional(rollbackFor = Exception.class) ConsumerRegisterResponse register(ConsumerRegisterRequest request); @Override @Transactional(rollbackFor = Exception.class) ConsumerDeRegisterResponse deRegister(ConsumerDeRegisterRequest deRegisterRequest); @Override void heartbeat(List<Long> ids); @Override List<ConsumerGroupConsumerEntity> getConsumerGroupByConsumerIds(List<Long> consumerIds); @Override List<ConsumerGroupConsumerEntity> getConsumerGroupByConsumerGroupIds(List<Long> consumerGroupIds); @Override //@Transactional(rollbackFor = Exception.class) ConsumerGroupRegisterResponse registerConsumerGroup(ConsumerGroupRegisterRequest request); @Override PublishMessageResponse publish(PublishMessageRequest request); void notifyClient(QueueEntity queueEntity); @Override PullDataResponse pullData(PullDataRequest request); @Override GetMessageCountResponse getMessageCount(GetMessageCountRequest request); @Override List<ConsumerEntity> findByHeartTimeInterval(long heartTimeInterval); @Override boolean deleteByConsumers(List<ConsumerEntity> consumers); @Override ConsumerEntity getConsumerByConsumerGroupId(Long consumerGroupId); @Override long countBy(Map<String, Object> conditionMap); @Override List<ConsumerEntity> getListBy(Map<String, Object> conditionMap); @Override FailMsgPublishAndUpdateResultResponse publishAndUpdateResultFailMsg(
            FailMsgPublishAndUpdateResultRequest request);  }### Answer:
@Test public void doSaveMsgTest() { List<Message01Entity> message01Entities=new ArrayList<Message01Entity>(); PublishMessageRequest request=new PublishMessageRequest(); PublishMessageResponse response=new PublishMessageResponse(); QueueEntity temp=new QueueEntity(); temp.setIp("fafd"); doThrow(new RuntimeException(new DataIntegrityViolationException(""))).when(message01Service).insertBatchDy(anyString(),anyString(),anyListOf(Message01Entity.class)); consumerServiceImpl.doSaveMsg(message01Entities, request, response, temp); assertEquals(true, consumerServiceImpl.dbFailMap.size()==0); reset(message01Service); doThrow(new RuntimeException("")).when(message01Service).insertBatchDy(anyString(),anyString(),anyListOf(Message01Entity.class)); try { consumerServiceImpl.doSaveMsg(message01Entities, request, response, temp); } catch (Exception e) { } assertEquals(true, consumerServiceImpl.dbFailMap.size()>0); }