@Test public void mergeSequenceSegmentTest() { List<Position> list = new ArrayList<>(); Position position = new Position(1, 1, 1, 1); Position position2 = new Position(3, 3, 3, 3); Position position3 = new Position(4, 4, 4, 4); Position position4 = new Position(5, 5, 5, 5); Position position5 = new Position(6, 6, 6, 6); list.add(position3); list.add(position); list.add(position2); list.add(position5); list.add(position4); List<Position> positions = sortByAckStartIndex(list); List<Position> positions1 = mergeSequenceSegment(positions); System.out.println(positions1); }
private List<Position> mergeSequenceSegment(List<Position> sortPositionList) { if (sortPositionList.size() <= 1) { return sortPositionList; } List<Position> mergeList = new ArrayList<>(); for (int i = 0; i < sortPositionList.size(); i++) { Position current = sortPositionList.get(i); Position last = current; int skipIndex = 0; for (int j = i + 1; j < sortPositionList.size(); j++) { Position next = sortPositionList.get(j); Position mergedPosition = tryMergeTwoSegment(last, next); if (mergedPosition != null) { last = mergedPosition; skipIndex ++; } else { break; } } mergeList.add(last); i += skipIndex; } return mergeList; }
ConcurrentConsumption extends Service implements ConcurrentConsumer { private List<Position> mergeSequenceSegment(List<Position> sortPositionList) { if (sortPositionList.size() <= 1) { return sortPositionList; } List<Position> mergeList = new ArrayList<>(); for (int i = 0; i < sortPositionList.size(); i++) { Position current = sortPositionList.get(i); Position last = current; int skipIndex = 0; for (int j = i + 1; j < sortPositionList.size(); j++) { Position next = sortPositionList.get(j); Position mergedPosition = tryMergeTwoSegment(last, next); if (mergedPosition != null) { last = mergedPosition; skipIndex ++; } else { break; } } mergeList.add(last); i += skipIndex; } return mergeList; } }
ConcurrentConsumption extends Service implements ConcurrentConsumer { private List<Position> mergeSequenceSegment(List<Position> sortPositionList) { if (sortPositionList.size() <= 1) { return sortPositionList; } List<Position> mergeList = new ArrayList<>(); for (int i = 0; i < sortPositionList.size(); i++) { Position current = sortPositionList.get(i); Position last = current; int skipIndex = 0; for (int j = i + 1; j < sortPositionList.size(); j++) { Position next = sortPositionList.get(j); Position mergedPosition = tryMergeTwoSegment(last, next); if (mergedPosition != null) { last = mergedPosition; skipIndex ++; } else { break; } } mergeList.add(last); i += skipIndex; } return mergeList; } ConcurrentConsumption(ClusterManager clusterManager, StoreService storeService, PartitionManager partitionManager,
                          MessageRetry messageRetry, PositionManager positionManager,
                          FilterMessageSupport filterMessageSupport, ArchiveManager archiveManager, SessionManager sessionManager); }
ConcurrentConsumption extends Service implements ConcurrentConsumer { private List<Position> mergeSequenceSegment(List<Position> sortPositionList) { if (sortPositionList.size() <= 1) { return sortPositionList; } List<Position> mergeList = new ArrayList<>(); for (int i = 0; i < sortPositionList.size(); i++) { Position current = sortPositionList.get(i); Position last = current; int skipIndex = 0; for (int j = i + 1; j < sortPositionList.size(); j++) { Position next = sortPositionList.get(j); Position mergedPosition = tryMergeTwoSegment(last, next); if (mergedPosition != null) { last = mergedPosition; skipIndex ++; } else { break; } } mergeList.add(last); i += skipIndex; } return mergeList; } ConcurrentConsumption(ClusterManager clusterManager, StoreService storeService, PartitionManager partitionManager,
                          MessageRetry messageRetry, PositionManager positionManager,
                          FilterMessageSupport filterMessageSupport, ArchiveManager archiveManager, SessionManager sessionManager); @Override PullResult getMessage(Consumer consumer, int count, long ackTimeout, long accessTimes, int concurrent); @Override boolean acknowledge(MessageLocation[] locations, Consumer consumer, boolean isSuccessAck); }
ConcurrentConsumption extends Service implements ConcurrentConsumer { private List<Position> mergeSequenceSegment(List<Position> sortPositionList) { if (sortPositionList.size() <= 1) { return sortPositionList; } List<Position> mergeList = new ArrayList<>(); for (int i = 0; i < sortPositionList.size(); i++) { Position current = sortPositionList.get(i); Position last = current; int skipIndex = 0; for (int j = i + 1; j < sortPositionList.size(); j++) { Position next = sortPositionList.get(j); Position mergedPosition = tryMergeTwoSegment(last, next); if (mergedPosition != null) { last = mergedPosition; skipIndex ++; } else { break; } } mergeList.add(last); i += skipIndex; } return mergeList; } ConcurrentConsumption(ClusterManager clusterManager, StoreService storeService, PartitionManager partitionManager,
                          MessageRetry messageRetry, PositionManager positionManager,
                          FilterMessageSupport filterMessageSupport, ArchiveManager archiveManager, SessionManager sessionManager); @Override PullResult getMessage(Consumer consumer, int count, long ackTimeout, long accessTimes, int concurrent); @Override boolean acknowledge(MessageLocation[] locations, Consumer consumer, boolean isSuccessAck);  }