VPF::Сделать из однопоточного кода многопоточный

Temdegon

Дата 15.1.2009, 05:11 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 429
Регистрация: 11.10.2008
Где: Minsk

Репутация: 7
Всего: 9

Кое-как написал код для поиска рабочих proxy-серверов из списка в файле.
Но программа работает очень медленно из-за того, что для проверки серверов используется только один поток.
Помогите пожалуйста переделать этот код так, что бы серверы проверялись в несколько потоков параллельно.
Нужно как-то сделать, что бы вместо вызова метода checkProxy создавался и стартовался новый поток, который бы делал то же самое, что и этот метод. Потом нужно как-то контролировать, сколько потоков уже работает, что бы не создать их слишком много.
Есть еще одна вещь, которую я просто взял из FAQ (спасибо Platon). При работе с HttpClient, в трех местах задается таймаут. Но за что конкретно отвечает каждый из них мне непонятно. Обьясните плиз. И в каком случае мы получим IOException в методе checkProxy?
А еще хотелось бы услышать любые замечания по этому коду. По улучшению производительности, по оформлению и т.д.

Код


import org.apache.commons.httpclient.*;
import org.apache.commons.httpclient.methods.*;
import org.apache.commons.httpclient.cookie.CookiePolicy;
import java.io.*;
import org.apache.commons.httpclient.params.HttpMethodParams;
import org.apache.commons.io.IOUtils;

public class ProxyChecker {
    //private static final String DOMAIN = "sharkscope.com";
    private static final String SITE_ROOT = "http://ya.ru/";
    private static final String DEFAULT_CHARSET = "UTF-8";
    private static final String SITE_CHARSET = "CP1251";
    private static final String FIND_STRING = "Сделать стартовой";
    //private static final String FIND_STRING = "Not Logged In. Searches Remaining";
    private HttpClient httpClient;
    String file;
    OutputStream osGood;
    OutputStream osBad;
    HostConfiguration host;
    public ProxyChecker(){
        httpClient = new HttpClient();
        httpClient.getParams().setContentCharset(DEFAULT_CHARSET);
        httpClient.getParams().setCookiePolicy(CookiePolicy.BROWSER_COMPATIBILITY);
        file = null;
        try {
            InputStream is = new FileInputStream("ReadyProxys.txt");
            osGood = new FileOutputStream("GoodProxies.txt");
            osBad = new FileOutputStream("BadProxies.txt");
            file = IOUtils.toString(is);
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
    public static void main(String[] args) throws UnsupportedEncodingException {
        ProxyChecker p = new ProxyChecker();
        p.start();
    }
    private boolean checkProxy(String proxyStr, int num){
        System.out.println("(" + num + ") " + "Checking proxy: " + proxyStr);
        host = new HostConfiguration();
        String[] p = proxyStr.split(":");
        int port = Integer.parseInt(p[1]);
        if (p[0].length()>8 && port > 0){
            host.setProxy(p[0], port);
        }else{
            System.out.println("BAD PROXY0");
            return false;
        }
        httpClient.setHostConfiguration(host);
        HttpState state = new HttpState();
        httpClient.setState(state);
        httpClient.getParams().setSoTimeout(10000);
        GetMethod getMethod = new GetMethod(SITE_ROOT);
        getMethod.getParams().setContentCharset(SITE_CHARSET);
        getMethod.getParams().setSoTimeout(10000);
        httpClient.getParams().setParameter(HttpMethodParams.RETRY_HANDLER, new DefaultHttpMethodRetryHandler(2, false));
        httpClient.getHttpConnectionManager().getParams().setConnectionTimeout(20000);
        try {
            int result = httpClient.executeMethod(getMethod);
            if (result == HttpStatus.SC_OK) {
                String testStr = getMethod.getResponseBodyAsString();
                if (testStr.indexOf(FIND_STRING) != -1){
                    System.out.println("GOOD PROXY");
                    return true;
                }else{
                    System.out.println("BAD PROXY1");
                    return false;
                }
            } else {
                System.out.println("BAD PROXY2");
                return false;
            }
        } catch (IOException e) {
            System.out.println("BAD PROXY3");
            return false;
        } finally {
            getMethod.releaseConnection();
        }
    }

    private void start() {
        String[] allProxies = file.split("\n");
        for(int i = 0; i < allProxies.length ; i++){
            if (checkProxy(allProxies[i], i)){
                try {
                    IOUtils.write(allProxies[i]+"\n", osGood);
                    osGood.flush();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }else{
                try {
                    IOUtils.write(allProxies[i]+"\n", osBad);
                    osBad.flush();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }

}

LSD

Дата 15.1.2009, 13:06 (ссылка)

(нет голосов)

Загрузка ...

Leprechaun Software Developer

Профиль
Группа: Модератор
Сообщений: 15718
Регистрация: 24.3.2004
Где: Dublin

Репутация: 210
Всего: 538

Цитата(Temdegon @ 15.1.2009, 05:11

)

А еще хотелось бы услышать любые замечания по этому коду. По улучшению производительности, по оформлению и т.д.

Создаешь ThreadPoolExecutor и запускаешь его. Затем читаешь список прокси и для каждого прокси сервера создаешь новый таск (какой нибудь Callable<ProxyCheckResult>). Все Future<ProxyCheckResult> складываешь в лист.
После того как засабмитил все прокси, идешь по списку и получаешь результат выполнения для каждого Future<ProxyCheckResult> и выводишь его куда требуется.

--------------------

Disclaimer: this post contains explicit depictions of personal opinion. So, if it sounds sarcastic, don't take it seriously. If it sounds dangerous, do not try this at home or at all. And if it offends you, just don't read it.

AntonSaburov

Дата 15.1.2009, 13:13 (ссылка)

(нет голосов)

Загрузка ...

Штурман

Профиль
Группа: Модератор
Сообщений: 5658
Регистрация: 2.7.2002
Где: Санкт-Петербург

Репутация: 51
Всего: 118

Вообщем-то совсем не проблема выделить нужный метод в отдельный класс, который либо наследуется от Thread, либо реализцет интерфейс Runnable. Тогда им можно управлять, смотреть его состояние и прочая - см. Thread

По поводу ограничения - можно использовать Semaphore - описание есть в API Doc.
Ну или тут приличное описание - http://java.sun.com/developer/technicalArt...SE/concurrency/

--------------------

Мои проекты для Винграда

Temdegon

Дата 16.1.2009, 05:30 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 429
Регистрация: 11.10.2008
Где: Minsk

Репутация: 7
Всего: 9

Что-то я не совсем понял с этими ThreadPoolExecutor, Callable и Future. Самих адресов, которые мне нужно проверить может быть очень много - тысячи, а может и десятки тысяч, т.к. прога сама будет их собирать по инету. Но одновременно должно выполняться некое фиксированное число потоков, которое можно было бы выставлять, в зависимости от пропускной способности канала - я думаю, штук 20-50. Не совсем врубился, какое из чисел, передаваемых в конструктор ThreadPoolExecutor за это отвечает. Еще меня смущает тот факт, что будет создаваться и висеть в памяти очень много обьектов Callable и Future. Не будет ли это поводом для OutOfMemory?
В главном классе код запуска выглядит примерно так:

Код


        threadPool = new ThreadPoolExecutor(10, 100, 10000, TimeUnit.SECONDS, queue);
        String[] allProxies = file.split("\n");
        for(int i = 0; i < allProxies.length ; i++){
            threadPool.submit(new ProxyTask(allProxies[i], i));
        }

А сам Callable вот так:

Код


public class ProxyTask implements Callable<Result>{
    private String ip;
........................
    private int num;

    public ProxyTask(String ip, int num){
        this.ip = ip;
        this.num = num;
    }
    public Result call() throws Exception {
        return new Result(checkProxy(ip, num), ip, num);
    }
    private boolean checkProxy(String proxyStr, int num){
...................
   }
}

Result

Код


public class Result implements Future{
    private int num;
    private String address;
    private boolean resutl;

    Result(boolean result, String str, int num) {
        this.resutl = result;
        this.address = str;
        this.num = num;
    }
............................................
     пять  методов от интерфейса, что делать в которых я не знаю
............................................
}

new ThreadPoolExecutor(10, 100, 10000, TimeUnit.SECONDS, queue); - что здесь означают параметры? из документации я ничего толком не понял.
Как узнать, что все потоки завершились?
Как правильно завершить Executor?

LSD

Дата 16.1.2009, 17:40 (ссылка)

(нет голосов)

Загрузка ...

Leprechaun Software Developer

Профиль
Группа: Модератор
Сообщений: 15718
Регистрация: 24.3.2004
Где: Dublin

Репутация: 210
Всего: 538

ThreadPoolExecutor-у надо задать максимальное количество потоков которые будут проверять прокси. Самому создавать Future не надо, их создаст и вернет тебе ThreadPoolExecutor когда ты засабмитишь ему Callable. У Future есть два метода чтобы получить результат: один ждет бесконечно, второй определённое время.

Как-то так это должно выглядеть.

Код


public class Test
{
  public static final int MAX_THREAD_COUNT = 10;

  public static void main(String[] args) throws Exception
  {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(1, MAX_THREAD_COUNT, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>());
    ArrayList<Future<ProxyCheckResult>> results = new ArrayList<Future<ProxyCheckResult>>(512);
    try
    {
      BufferedReader reader = new BufferedReader(new FileReader("proxy.txt"));
      String str;
      while((str = reader.readLine()) != null)
      {
        results.add(executor.submit(new ProxyCheck(str)));
      }
      reader.close();

      for(Future<ProxyCheckResult> resultFuture : results)
      {
        final ProxyCheckResult proxyCheckResult = resultFuture.get();
        System.out.println("proxyCheckResult = " + proxyCheckResult);
      }
    }
    finally
    {
      executor.shutdownNow();
    }
  }

  public static class ProxyCheckResult
  {
    private int num;
    private String address;
    private boolean resutl;
  }

  private static class ProxyCheck implements Callable<ProxyCheckResult>
  {
    private String proxy;

    private ProxyCheck(String proxy)
    {
      this.proxy = proxy;
    }

    public ProxyCheckResult call() throws Exception
    {
      return null;//check it and return result
    }
  }
}

(конечно это просто пример, а не то как должен выглядеть реальный код)

--------------------

Temdegon

Дата 16.1.2009, 18:23 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 429
Регистрация: 11.10.2008
Где: Minsk

Репутация: 7
Всего: 9

Спасибо LSD, буду разбираться. А за что отвечает первое число в конструкторе?
Я так понимаю, сюда

Код


      for(Future<ProxyCheckResult> resultFuture : results)
      {
        final ProxyCheckResult proxyCheckResult = resultFuture.get();
        System.out.println("proxyCheckResult = " + proxyCheckResult);
      }

зайдет только после того, как потоки отработают?

LSD

Дата 16.1.2009, 18:59 (ссылка)

(нет голосов)

Загрузка ...

Leprechaun Software Developer

Профиль
Группа: Модератор
Сообщений: 15718
Регистрация: 24.3.2004
Где: Dublin

Репутация: 210
Всего: 538

Цитата(Temdegon @ 16.1.2009, 18:23

)

А за что отвечает первое число в конструкторе?

Умолчальное количество потоков. После создания в пуле будет именно такое количество потоков. Параметр имеет значение для пулов которые могут быть нагружены работой частично. В этой задаче этот параметр не важен.

Цитата(Temdegon @ 16.1.2009, 18:23

)

Я так понимаю, сюда

зайдет только после того, как потоки отработают?

Нет. Сюда он зайдет после того как засабмитит все таски. Future.get() блокируемый метод, он ждет пока не будет готов результат.

--------------------

Temdegon

Дата 16.1.2009, 19:09 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 429
Регистрация: 11.10.2008
Где: Minsk

Репутация: 7
Всего: 9

Еще раз спасибо. Основные моменты вроди бы прояснились, и я думаю, что все получится

ivg

Дата 17.1.2009, 15:00 (ссылка)

(нет голосов)

Загрузка ...

Autonomous R&D

Профиль
Группа: Участник
Сообщений: 686
Регистрация: 8.2.2006
Где: Екатеринбург

Репутация: 33
Всего: 81

Цитата(LSD @ 16.1.2009, 20:59

)

После создания в пуле будет именно такое количество потоков

Небольшое уточнение: При вызове конструктора, эти потоки не создаются. Они создаются в первые n-вызовов методов, предназначенных для постановки заданий. Чтобы потоки ко времени постановки заданий уже были бы созданы, нужно явно вызвать ThreadPoolExecutor#setCorePoolSize(int corePoolSize);

LSD

Дата 20.1.2009, 12:47 (ссылка)

(нет голосов)

Загрузка ...

Leprechaun Software Developer

Профиль
Группа: Модератор
Сообщений: 15718
Регистрация: 24.3.2004
Где: Dublin

Репутация: 210
Всего: 538

Цитата(ivg @ 17.1.2009, 15:00

)

Чтобы потоки ко времени постановки заданий уже были бы созданы, нужно явно вызвать ThreadPoolExecutor#setCorePoolSize(int corePoolSize);

Тогда уж лучше prestartAllCoreThreads() или prestartCoreThread() если нужно стартовать один поток.

--------------------

0 Пользователей читают эту тему (0 Гостей и 0 Скрытых Пользователей)
0 Пользователей:
« Предыдущая тема \| Java: Общие вопросы \| Следующая тема »