Демо-проект: Apache CXF, Jetty, MySQL

Решил выложить на Github небольшой демо-проект: клиент-серверное приложение, управляющее аккаунтами (счетами).

Начальная задача: сделать сервер, который позволяет читать состояние счета по его ID, а также добавлять/снимать средства со счета. Состояния счетов хранятся с БД. Сервер должен работать в нагруженное среде, поэтому нужно добавить кеширование. Транспортный протокол может быть любой. Я выбрал веб-сервисы, реализация веб-сервисов была сделана с помощью Apache CXF. После запуска сервера WSDL можно посмотреть по адресу: http://localhost:9000/accountService?wsdl

Вторая часть задачки: сделать клиента, который в несколько потоков читает/пишет значения счетов. Кол-во читателей/писателей, а также множество идентификаторов счетов задается параметрами клиента (например, из командной строки).

Третья часть задачки: сервер должен статистику работы скидывать в файл. Статистика должна быть такой: общее кол-во обработанных запросов, а также удельное кол-во запросов в ед. времени (т.е. rps - requests per second). Нужно предусмотреть возможность сброса статсы в ноль. Сброс статсы в ноль я сделал с помощью вызова HTTP-хендлера, который и обрабатывается Jetty. Сброс статсы осуществляется HTTP GET-ом урлы http://localhost:9001/zero. По урлу http://localhost:9001/getStats получаем текущее значение rps.

Несколько моментов про реализацию. Итак, рассмотрим реализацию AccountServiceImpl:

@WebService(endpointInterface = "net.iryndin.clientserverdemo1.commons.api.AccountService",serviceName = "AccountService")
public class AccountServiceImpl implements AccountService {
    private Map cache = new ConcurrentHashMap<>();

    @Override
    public Long getAmount(Integer id) {
        AtomicLong balance = cache.get(id);
        statsHelper.incrQueries();
        if (balance == null) {
            return 0L;
        } else {
            return balance.get();
        }
    }

    @Override
    public void addAmount(final Integer id, Long value) throws Exception {
        AtomicLong balance = cache.get(id);
        if (balance == null) {
            balance = new AtomicLong(0);
            cache.put(id, balance);
        }
        balance.addAndGet(value);
        final AtomicLong finalBalance = balance;
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    accountDao.save(id, finalBalance.get());
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        });
        statsHelper.incrQueries();
    }
}  

Видно, что чтение и запись идет прямиком к кеш, т.е. скорость чтения и записи определяется скоростью работы кеша. Запись в базу запускается в отдельном потоке. Если бы она запускалась из этого же потока, то и производительность записи была бы совсем другой. На самом деле, такое решение, конечно, не всегда приемлемо, но поскольку это тестовый проект, я сделал именно так.

Теперь рассмотрим моменты реализации клиента. См. файл MainClient:

    private static void runReaders(int count) {
        if (count <=0) {
            System.out.println("No readers run");
            return;
        }
        ExecutorService executorService = Executors.newFixedThreadPool(count);
        System.out.println("Run readers: " + count);
        while (count > 0) {
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    while (running) {
                        accountService.getAmount(idsHelper.getRandomId());
                    }
                }
            });
            count--;
        }
    }

    private static void runWriters(int count) {
        if (count <=0) {
            System.out.println("No writers run");
            return;
        }
        ExecutorService executorService = Executors.newFixedThreadPool(count);
        System.out.println("Run writers: " + count);
        while (count > 0) {
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    while (running) {
                        boolean positive = new Random().nextBoolean();
                        int balance = new Random().nextInt(100000);
                        balance = positive ? balance : -balance;
                        try {
                            accountService.addAmount(idsHelper.getRandomId(), (long) balance);
                        } catch (Exception e) {
                            //
                        }
                    }
                }
            });
            count--;
        }
    }

Мы запускаем потоки по числу rCount, wCount и крутимся там в бесконечном цикле (в бесконечном, так как переменную running в коде никто не меняет). Все этот создает неплохую нагрузку на сервер.

Последнее, рассмотрим скрипт на Python 3 (measure.py), который запускает клиент с различными значениями параметров rCount, wCount и измеряет производительность (просто считывает данные с урла http://localhost:9001/getStats):

import subprocess
import time
import urllib.request

def main():
    readers = [0, 20, 40, 60, 80, 100]
    writers = [0, 20, 40, 60, 80, 100]

    output = open('rps.txt','w', 1)
    
    for r in readers:
        for w in writers:
            run_client(r,w,output)
            print('Done reader=%d, writer=%d' % (r,w))
            
    output.close()

def run_client(readers_qty, writers_qty, output):        
    if readers_qty==0 and writers_qty==0:
        return
    proc = subprocess.Popen(['java', '-jar', '-Xms2G', '-Xmx8G', 'client.jar', str(readers_qty), str(writers_qty)], shell=False)
    time.sleep(1)
    f = urllib.request.urlopen('http://localhost:9001/zero')
    time.sleep(30)
    f = urllib.request.urlopen('http://localhost:9001/getStats')
    s = f.read()
    rps = int(s)
    print('rps=%d' % rps)
    output.write('(%d,%d): %d\n' % (readers_qty,writers_qty,rps))
    
    proc.terminate()
    time.sleep(5)
     
if __name__ == "__main__":
    main()

В результате, мы получаем следующую табличку (по горизонтали: число wCount, по вертикали: rCount, значения в ячейках: requests-per-second):

rCount/wCount	0	20	40	60	80	100
0	-	12197	12319	12865	12648	13614
20	13324	13240	13049	12657	12499	12858
40	13198	12249	12676	13112	12882	12442
60	12420	12250	12348	12928	12534	13013
80	12531	12934	12304	12428	13133	12978
100	12972	13204	11869	11541	12781	12496

Т.е. сервер держит где-то 12-13К реквестов. Это число полносью определяется производительностью ConcurrentHashMap.